- Repo-Struktur für Backend, Frontend, Dokumentation und Gitea-Workflows - FastAPI-Skeleton mit pydantic-settings und RFC-7807-artigem Fehlerformat - Vollständiges Datenmodell (15 Tabellen) als SQLAlchemy-2.0-Modelle - Alembic gegen die Async-Engine inklusive Erstmigration - Idempotenter Seed für den deutschen Kategoriebaum und Standardregeln - Endpunkte /api/health und /api/version - pytest-Infrastruktur mit transaktionsisolierten Fixtures Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014e7t8UpmoVNMtWivY5LiSH
109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
"""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
|