"""RFC 7807 problem+json errors for every failure the API returns.""" from __future__ import annotations from collections.abc import Mapping from typing import Any from fastapi import FastAPI, Request, status from fastapi.exceptions import HTTPException, RequestValidationError from fastapi.responses import JSONResponse PROBLEM_TYPE = "application/problem+json" class Problem(Exception): def __init__( self, status_code: int, title: str, detail: str | None = None, *, extra: dict[str, Any] | None = None, headers: Mapping[str, str] | None = None, ) -> None: self.status_code = status_code self.title = title self.detail = detail self.extra = extra or {} self.headers = headers def _response( status_code: int, title: str, detail: str | None = None, extra: dict[str, Any] | None = None, headers: Mapping[str, str] | None = None, ) -> JSONResponse: body: dict[str, Any] = {"status": status_code, "title": title} if detail: body["detail"] = detail if extra: body.update(extra) return JSONResponse(body, status_code=status_code, media_type=PROBLEM_TYPE, headers=headers) def install_error_handlers(app: FastAPI) -> None: @app.exception_handler(Problem) async def _problem(_: Request, exc: Problem) -> JSONResponse: return _response(exc.status_code, exc.title, exc.detail, exc.extra, exc.headers) @app.exception_handler(HTTPException) async def _http(_: Request, exc: HTTPException) -> JSONResponse: detail = exc.detail if isinstance(exc.detail, str) else None return _response(exc.status_code, _title_for(exc.status_code), detail, headers=exc.headers) @app.exception_handler(RequestValidationError) async def _validation(_: Request, exc: RequestValidationError) -> JSONResponse: return _response( status.HTTP_422_UNPROCESSABLE_ENTITY, "Validation error", extra={"errors": exc.errors()}, ) def _title_for(code: int) -> str: return { 400: "Bad request", 401: "Unauthorized", 403: "Forbidden", 404: "Not found", 409: "Conflict", 429: "Too many requests", }.get(code, "Error")