"""Every API error is JSON `{"detail": "", "code": ""}`, with extra fields when useful.""" from __future__ import annotations from typing import Any from fastapi import FastAPI, Request from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from starlette.exceptions import HTTPException as StarletteHTTPException HTTP_CODES = { 400: "bad_request", 401: "no_session", 403: "forbidden", 404: "not_found", 405: "method_not_allowed", 409: "conflict", 413: "too_large", 429: "rate_limited", 503: "unavailable", } class ApiError(Exception): def __init__( self, status: int, code: str, detail: str, *, headers: dict[str, str] | None = None, extra: dict[str, Any] | None = None, ): super().__init__(detail) self.status = status self.code = code self.detail = detail self.headers = headers or {} self.extra = extra or {} def body(self) -> dict[str, Any]: return {"detail": self.detail, "code": self.code, **self.extra} def error_response(err: ApiError) -> JSONResponse: return JSONResponse(err.body(), status_code=err.status, headers=err.headers) def install(app: FastAPI) -> None: @app.exception_handler(ApiError) async def _api_error(_: Request, exc: ApiError) -> JSONResponse: return error_response(exc) @app.exception_handler(RequestValidationError) async def _invalid(_: Request, exc: RequestValidationError) -> JSONResponse: errors = [{"loc": [str(x) for x in e.get("loc", ())], "msg": str(e.get("msg", ""))} for e in exc.errors()] where = ", ".join(".".join(e["loc"][1:]) or e["loc"][0] for e in errors if e["loc"]) or "request" return JSONResponse( {"detail": f"Invalid request: {where}", "code": "invalid_request", "errors": errors}, status_code=422 ) @app.exception_handler(StarletteHTTPException) async def _http(_: Request, exc: StarletteHTTPException) -> JSONResponse: code = HTTP_CODES.get(exc.status_code, "http_error") return JSONResponse( {"detail": str(exc.detail), "code": code}, status_code=exc.status_code, headers=exc.headers or None )