Spaces:
Paused
Paused
| 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 | |