exceptions.py 7.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256
  1. from collections.abc import Mapping, Sequence
  2. from typing import Annotated, Any, TypedDict
  3. from annotated_doc import Doc
  4. from pydantic import BaseModel, create_model
  5. from starlette.exceptions import HTTPException as StarletteHTTPException
  6. from starlette.exceptions import WebSocketException as StarletteWebSocketException
  7. class EndpointContext(TypedDict, total=False):
  8. function: str
  9. path: str
  10. file: str
  11. line: int
  12. class HTTPException(StarletteHTTPException):
  13. """
  14. An HTTP exception you can raise in your own code to show errors to the client.
  15. This is for client errors, invalid authentication, invalid data, etc. Not for server
  16. errors in your code.
  17. Read more about it in the
  18. [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/).
  19. ## Example
  20. ```python
  21. from fastapi import FastAPI, HTTPException
  22. app = FastAPI()
  23. items = {"foo": "The Foo Wrestlers"}
  24. @app.get("/items/{item_id}")
  25. async def read_item(item_id: str):
  26. if item_id not in items:
  27. raise HTTPException(status_code=404, detail="Item not found")
  28. return {"item": items[item_id]}
  29. ```
  30. """
  31. def __init__(
  32. self,
  33. status_code: Annotated[
  34. int,
  35. Doc(
  36. """
  37. HTTP status code to send to the client.
  38. Read more about it in the
  39. [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/#use-httpexception)
  40. """
  41. ),
  42. ],
  43. detail: Annotated[
  44. Any,
  45. Doc(
  46. """
  47. Any data to be sent to the client in the `detail` key of the JSON
  48. response.
  49. Read more about it in the
  50. [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/#use-httpexception)
  51. """
  52. ),
  53. ] = None,
  54. headers: Annotated[
  55. Mapping[str, str] | None,
  56. Doc(
  57. """
  58. Any headers to send to the client in the response.
  59. Read more about it in the
  60. [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/#add-custom-headers)
  61. """
  62. ),
  63. ] = None,
  64. ) -> None:
  65. super().__init__(status_code=status_code, detail=detail, headers=headers)
  66. class WebSocketException(StarletteWebSocketException):
  67. """
  68. A WebSocket exception you can raise in your own code to show errors to the client.
  69. This is for client errors, invalid authentication, invalid data, etc. Not for server
  70. errors in your code.
  71. Read more about it in the
  72. [FastAPI docs for WebSockets](https://fastapi.tiangolo.com/advanced/websockets/).
  73. ## Example
  74. ```python
  75. from typing import Annotated
  76. from fastapi import (
  77. Cookie,
  78. FastAPI,
  79. WebSocket,
  80. WebSocketException,
  81. status,
  82. )
  83. app = FastAPI()
  84. @app.websocket("/items/{item_id}/ws")
  85. async def websocket_endpoint(
  86. *,
  87. websocket: WebSocket,
  88. session: Annotated[str | None, Cookie()] = None,
  89. item_id: str,
  90. ):
  91. if session is None:
  92. raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
  93. await websocket.accept()
  94. while True:
  95. data = await websocket.receive_text()
  96. await websocket.send_text(f"Session cookie is: {session}")
  97. await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}")
  98. ```
  99. """
  100. def __init__(
  101. self,
  102. code: Annotated[
  103. int,
  104. Doc(
  105. """
  106. A closing code from the
  107. [valid codes defined in the specification](https://datatracker.ietf.org/doc/html/rfc6455#section-7.4.1).
  108. """
  109. ),
  110. ],
  111. reason: Annotated[
  112. str | None,
  113. Doc(
  114. """
  115. The reason to close the WebSocket connection.
  116. It is UTF-8-encoded data. The interpretation of the reason is up to the
  117. application, it is not specified by the WebSocket specification.
  118. It could contain text that could be human-readable or interpretable
  119. by the client code, etc.
  120. """
  121. ),
  122. ] = None,
  123. ) -> None:
  124. super().__init__(code=code, reason=reason)
  125. RequestErrorModel: type[BaseModel] = create_model("Request")
  126. WebSocketErrorModel: type[BaseModel] = create_model("WebSocket")
  127. class FastAPIError(RuntimeError):
  128. """
  129. A generic, FastAPI-specific error.
  130. """
  131. class DependencyScopeError(FastAPIError):
  132. """
  133. A dependency declared that it depends on another dependency with an invalid
  134. (narrower) scope.
  135. """
  136. class ValidationException(Exception):
  137. def __init__(
  138. self,
  139. errors: Sequence[Any],
  140. *,
  141. endpoint_ctx: EndpointContext | None = None,
  142. ) -> None:
  143. self._errors = errors
  144. self.endpoint_ctx = endpoint_ctx
  145. ctx = endpoint_ctx or {}
  146. self.endpoint_function = ctx.get("function")
  147. self.endpoint_path = ctx.get("path")
  148. self.endpoint_file = ctx.get("file")
  149. self.endpoint_line = ctx.get("line")
  150. def errors(self) -> Sequence[Any]:
  151. return self._errors
  152. def _format_endpoint_context(self) -> str:
  153. if not (self.endpoint_file and self.endpoint_line and self.endpoint_function):
  154. if self.endpoint_path:
  155. return f"\n Endpoint: {self.endpoint_path}"
  156. return ""
  157. context = f'\n File "{self.endpoint_file}", line {self.endpoint_line}, in {self.endpoint_function}'
  158. if self.endpoint_path:
  159. context += f"\n {self.endpoint_path}"
  160. return context
  161. def __str__(self) -> str:
  162. message = f"{len(self._errors)} validation error{'s' if len(self._errors) != 1 else ''}:\n"
  163. for err in self._errors:
  164. message += f" {err}\n"
  165. message += self._format_endpoint_context()
  166. return message.rstrip()
  167. class RequestValidationError(ValidationException):
  168. def __init__(
  169. self,
  170. errors: Sequence[Any],
  171. *,
  172. body: Any = None,
  173. endpoint_ctx: EndpointContext | None = None,
  174. ) -> None:
  175. super().__init__(errors, endpoint_ctx=endpoint_ctx)
  176. self.body = body
  177. class WebSocketRequestValidationError(ValidationException):
  178. def __init__(
  179. self,
  180. errors: Sequence[Any],
  181. *,
  182. endpoint_ctx: EndpointContext | None = None,
  183. ) -> None:
  184. super().__init__(errors, endpoint_ctx=endpoint_ctx)
  185. class ResponseValidationError(ValidationException):
  186. def __init__(
  187. self,
  188. errors: Sequence[Any],
  189. *,
  190. body: Any = None,
  191. endpoint_ctx: EndpointContext | None = None,
  192. ) -> None:
  193. super().__init__(errors, endpoint_ctx=endpoint_ctx)
  194. self.body = body
  195. class PydanticV1NotSupportedError(FastAPIError):
  196. """
  197. A pydantic.v1 model is used, which is no longer supported.
  198. """
  199. class FastAPIDeprecationWarning(UserWarning):
  200. """
  201. A custom deprecation warning as DeprecationWarning is ignored
  202. Ref: https://sethmlarson.dev/deprecations-via-warnings-dont-work-for-python-libraries
  203. """