-
Notifications
You must be signed in to change notification settings - Fork 2
/
helpers.py
56 lines (45 loc) · 2.15 KB
/
helpers.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
from __future__ import annotations
from typing import Any, Type
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
from .routing import SchemaAPIRoute
from .exceptions import BaseGenericHTTPException
from .interfaces import AbstractResponseSchema
from ._compat import model_to_dict
def wrap_error_responses(app: FastAPI, error_response_schema: Type[AbstractResponseSchema]) -> FastAPI:
"""Wraps all exception handlers with the provided response schema.
Args:
app (FastAPI): A FastAPI application instance.
error_response_schema (Type[AbstractResponseSchema]): Response schema wrapper model.
Returns:
FastAPI: The application instance
"""
async def exception_handler(request, exc):
status_code = getattr(exc, "status_code") if not isinstance(exc, RequestValidationError) else 422
# due to: https://github.com/python/mypy/issues/12392 FIXME: when gets fixed
model = error_response_schema[Any] # type: ignore
return JSONResponse(
content=model_to_dict(model.from_exception_handler(request=request, exception=exc)),
status_code=status_code,
headers=getattr(exc, "headers", dict()),
)
app.add_exception_handler(RequestValidationError, exception_handler)
app.add_exception_handler(StarletteHTTPException, exception_handler)
app.add_exception_handler(BaseGenericHTTPException, exception_handler)
return app
def wrap_app_responses(app: FastAPI, route_class: Type[SchemaAPIRoute]) -> FastAPI:
"""Wraps all app defaults responses
Args:
app (FastAPI): A FastAPI application instance.
route_class (Type[SchemaAPIRoute]): The SchemaAPIRoute with your response schemas.
Returns:
FastAPI: The application instance.
"""
app.router.route_class = route_class
err_schema = getattr(route_class, "error_response_schema")
if err_schema is None:
err_schema = route_class.response_schema
app = wrap_error_responses(app, error_response_schema=err_schema)
return app