Aleksei Sokol
53f14a8624
All checks were successful
Run linters on applied template / Python 3.13 lint and build (push) Successful in 1m40s
Changes: - put ObservabilityMiddleware before ExceptionHandlerMiddleware to avoid repetative code - add application startup and last metrics update metrics along with CPU usage metric and threads count - move host and port to new uvicorn section at config along with new reload and forwarded_allow_ips - add request_id and remove trace_id/span_id generation if tracing is disabled - move logging logic from utils to observability - pass trace_id/span_id in HEX form
49 lines
2.0 KiB
Python
49 lines
2.0 KiB
Python
"""Mapper from exceptions to HTTP responses is defined here"""
|
|
|
|
from typing import Callable, Type
|
|
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.exceptions import HTTPException
|
|
|
|
|
|
class ExceptionMapper:
|
|
"""Maps exceptions to `JSONResponse` for FastAPI error handling."""
|
|
|
|
def __init__(self):
|
|
self._known_exceptions: dict[Type, Callable[[Exception], JSONResponse]] = {}
|
|
|
|
def register_simple(self, exception_type: Type, status_code: int, details: str) -> None:
|
|
"""Register simple response handler with setting status_code and details."""
|
|
self._known_exceptions[exception_type] = lambda _: JSONResponse(
|
|
{"error": f"{exception_type.__module__}.{exception_type.__qualname__}", "details": details},
|
|
status_code=status_code,
|
|
)
|
|
|
|
def register_func(self, exception_type: Type, func: Callable[[Exception], JSONResponse]) -> None:
|
|
"""Register complex response handler by passing function."""
|
|
self._known_exceptions[exception_type] = func
|
|
|
|
def get_status_code(self, exc: Exception) -> int:
|
|
"""Get status code of preparing response."""
|
|
if isinstance(exc, HTTPException):
|
|
return exc.status_code
|
|
if type(exc) in self._known_exceptions:
|
|
return self._known_exceptions[type(exc)](exc).status_code
|
|
return 500
|
|
|
|
def is_known(self, exc: Exception) -> bool:
|
|
|
|
return type(exc) in self._known_exceptions or isinstance(exc, HTTPException)
|
|
|
|
def apply(self, exc: Exception) -> JSONResponse:
|
|
if (res := self.apply_if_known(exc)) is not None:
|
|
return res
|
|
return JSONResponse({"code": 500, "detail": "unhandled exception"}, status_code=500)
|
|
|
|
def apply_if_known(self, exc: Exception) -> JSONResponse | None:
|
|
if isinstance(exc, HTTPException):
|
|
return JSONResponse({"code": exc.status_code, "detail": exc.detail}, status_code=exc.status_code)
|
|
if (t := type(exc)) in self._known_exceptions:
|
|
return self._known_exceptions[t](exc)
|
|
return None
|