| from fastapi.encoders import jsonable_encoder | |
| from fastapi.exceptions import RequestValidationError, WebSocketRequestValidationError | |
| from fastapi.utils import is_body_allowed_for_status_code | |
| from fastapi.websockets import WebSocket | |
| from starlette.exceptions import HTTPException | |
| from starlette.requests import Request | |
| from starlette.responses import JSONResponse, Response | |
| from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY, WS_1008_POLICY_VIOLATION | |
| async def http_exception_handler(request: Request, exc: HTTPException) -> Response: | |
| headers = getattr(exc, "headers", None) | |
| if not is_body_allowed_for_status_code(exc.status_code): | |
| return Response(status_code=exc.status_code, headers=headers) | |
| return JSONResponse( | |
| {"detail": exc.detail}, status_code=exc.status_code, headers=headers | |
| ) | |
| async def request_validation_exception_handler( | |
| request: Request, exc: RequestValidationError | |
| ) -> JSONResponse: | |
| return JSONResponse( | |
| status_code=HTTP_422_UNPROCESSABLE_ENTITY, | |
| content={"detail": jsonable_encoder(exc.errors())}, | |
| ) | |
| async def websocket_request_validation_exception_handler( | |
| websocket: WebSocket, exc: WebSocketRequestValidationError | |
| ) -> None: | |
| await websocket.close( | |
| code=WS_1008_POLICY_VIOLATION, reason=jsonable_encoder(exc.errors()) | |
| ) | |