| from typing import Any, Dict, Optional, Sequence, Type, Union |
|
|
| from pydantic import BaseModel, create_model |
| from starlette.exceptions import HTTPException as StarletteHTTPException |
| from starlette.exceptions import WebSocketException as StarletteWebSocketException |
| from typing_extensions import Annotated, Doc |
|
|
|
|
| class HTTPException(StarletteHTTPException): |
| """ |
| An HTTP exception you can raise in your own code to show errors to the client. |
| |
| This is for client errors, invalid authentication, invalid data, etc. Not for server |
| errors in your code. |
| |
| Read more about it in the |
| [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/). |
| |
| ## Example |
| |
| ```python |
| from fastapi import FastAPI, HTTPException |
| |
| app = FastAPI() |
| |
| items = {"foo": "The Foo Wrestlers"} |
| |
| |
| @app.get("/items/{item_id}") |
| async def read_item(item_id: str): |
| if item_id not in items: |
| raise HTTPException(status_code=404, detail="Item not found") |
| return {"item": items[item_id]} |
| ``` |
| """ |
|
|
| def __init__( |
| self, |
| status_code: Annotated[ |
| int, |
| Doc( |
| """ |
| HTTP status code to send to the client. |
| """ |
| ), |
| ], |
| detail: Annotated[ |
| Any, |
| Doc( |
| """ |
| Any data to be sent to the client in the `detail` key of the JSON |
| response. |
| """ |
| ), |
| ] = None, |
| headers: Annotated[ |
| Optional[Dict[str, str]], |
| Doc( |
| """ |
| Any headers to send to the client in the response. |
| """ |
| ), |
| ] = None, |
| ) -> None: |
| super().__init__(status_code=status_code, detail=detail, headers=headers) |
|
|
|
|
| class WebSocketException(StarletteWebSocketException): |
| """ |
| A WebSocket exception you can raise in your own code to show errors to the client. |
| |
| This is for client errors, invalid authentication, invalid data, etc. Not for server |
| errors in your code. |
| |
| Read more about it in the |
| [FastAPI docs for WebSockets](https://fastapi.tiangolo.com/advanced/websockets/). |
| |
| ## Example |
| |
| ```python |
| from typing import Annotated |
| |
| from fastapi import ( |
| Cookie, |
| FastAPI, |
| WebSocket, |
| WebSocketException, |
| status, |
| ) |
| |
| app = FastAPI() |
| |
| @app.websocket("/items/{item_id}/ws") |
| async def websocket_endpoint( |
| *, |
| websocket: WebSocket, |
| session: Annotated[str | None, Cookie()] = None, |
| item_id: str, |
| ): |
| if session is None: |
| raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION) |
| await websocket.accept() |
| while True: |
| data = await websocket.receive_text() |
| await websocket.send_text(f"Session cookie is: {session}") |
| await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}") |
| ``` |
| """ |
|
|
| def __init__( |
| self, |
| code: Annotated[ |
| int, |
| Doc( |
| """ |
| A closing code from the |
| [valid codes defined in the specification](https://datatracker.ietf.org/doc/html/rfc6455#section-7.4.1). |
| """ |
| ), |
| ], |
| reason: Annotated[ |
| Union[str, None], |
| Doc( |
| """ |
| The reason to close the WebSocket connection. |
| |
| It is UTF-8-encoded data. The interpretation of the reason is up to the |
| application, it is not specified by the WebSocket specification. |
| |
| It could contain text that could be human-readable or interpretable |
| by the client code, etc. |
| """ |
| ), |
| ] = None, |
| ) -> None: |
| super().__init__(code=code, reason=reason) |
|
|
|
|
| RequestErrorModel: Type[BaseModel] = create_model("Request") |
| WebSocketErrorModel: Type[BaseModel] = create_model("WebSocket") |
|
|
|
|
| class FastAPIError(RuntimeError): |
| """ |
| A generic, FastAPI-specific error. |
| """ |
|
|
|
|
| class ValidationException(Exception): |
| def __init__(self, errors: Sequence[Any]) -> None: |
| self._errors = errors |
|
|
| def errors(self) -> Sequence[Any]: |
| return self._errors |
|
|
|
|
| class RequestValidationError(ValidationException): |
| def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None: |
| super().__init__(errors) |
| self.body = body |
|
|
|
|
| class WebSocketRequestValidationError(ValidationException): |
| pass |
|
|
|
|
| class ResponseValidationError(ValidationException): |
| def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None: |
| super().__init__(errors) |
| self.body = body |
|
|
| def __str__(self) -> str: |
| message = f"{len(self._errors)} validation errors:\n" |
| for err in self._errors: |
| message += f" {err}\n" |
| return message |
|
|