"""Einheitliche Fehlerdarstellung im RFC-7807-Stil: {"detail": ..., "code": ...}.""" from typing import Any from fastapi import FastAPI, Request, status from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from starlette.exceptions import HTTPException as StarletteHTTPException # Starlette benennt die 422-Konstante gerade um – fester Wert vermeidet die Abhängigkeit. HTTP_422 = 422 class AppError(Exception): """Basisklasse für fachliche Fehler mit stabilem Fehlercode.""" status_code: int = status.HTTP_400_BAD_REQUEST code: str = "bad_request" def __init__(self, detail: str, *, code: str | None = None, status_code: int | None = None): super().__init__(detail) self.detail = detail if code is not None: self.code = code if status_code is not None: self.status_code = status_code class NotFoundError(AppError): status_code = status.HTTP_404_NOT_FOUND code = "not_found" class ConflictError(AppError): status_code = status.HTTP_409_CONFLICT code = "conflict" class ValidationError(AppError): status_code = HTTP_422 code = "validation_error" class AuthError(AppError): status_code = status.HTTP_401_UNAUTHORIZED code = "unauthorized" def problem( status_code: int, detail: str, code: str, extra: dict[str, Any] | None = None ) -> JSONResponse: body: dict[str, Any] = {"detail": detail, "code": code} if extra: body.update(extra) return JSONResponse(status_code=status_code, content=body) # Zuordnung der Standard-HTTP-Statuscodes auf sprechende Fehlercodes. _STATUS_CODES = { 400: "bad_request", 401: "unauthorized", 403: "forbidden", 404: "not_found", 405: "method_not_allowed", 409: "conflict", 413: "payload_too_large", 415: "unsupported_media_type", 422: "validation_error", 429: "too_many_requests", 500: "internal_error", } def register_exception_handlers(app: FastAPI) -> None: """Registriert alle Handler, damit jede Fehlerantwort dasselbe Format hat.""" @app.exception_handler(AppError) async def _app_error(_: Request, exc: AppError) -> JSONResponse: return problem(exc.status_code, exc.detail, exc.code) @app.exception_handler(StarletteHTTPException) async def _http_error(_: Request, exc: StarletteHTTPException) -> JSONResponse: code = _STATUS_CODES.get(exc.status_code, "error") detail = exc.detail if isinstance(exc.detail, str) else str(exc.detail) return problem(exc.status_code, detail, code) @app.exception_handler(RequestValidationError) async def _validation_error(_: Request, exc: RequestValidationError) -> JSONResponse: return problem( HTTP_422, "Die Anfrage enthält ungültige Felder.", "validation_error", {"errors": _serialise_errors(exc.errors())}, ) def _serialise_errors(errors: list[Any]) -> list[dict[str, Any]]: """Pydantic-Fehler auf JSON-serialisierbare Felder reduzieren.""" result = [] for error in errors: result.append( { "loc": [str(part) for part in error.get("loc", [])], "msg": error.get("msg", ""), "type": error.get("type", ""), } ) return result