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