Project started 🍾

app/api/errors.py

@@ -0,0 +1,50 @@
+"""Maps domain exceptions to HTTP responses. The only place that knows both."""
+
+from fastapi import FastAPI, Request, status
+from fastapi.responses import JSONResponse
+
+from app.core.logging import get_logger
+from app.domain.errors import (
+    AlreadyExistsError,
+    AuthenticationError,
+    ConflictError,
+    DependencyUnavailableError,
+    DomainError,
+    NotFoundError,
+    PermissionDeniedError,
+    ValidationError,
+)
+
+log = get_logger(__name__)
+
+_STATUS_BY_ERROR: dict[type[DomainError], int] = {
+    NotFoundError: status.HTTP_404_NOT_FOUND,
+    AlreadyExistsError: status.HTTP_409_CONFLICT,
+    ConflictError: status.HTTP_409_CONFLICT,
+    ValidationError: status.HTTP_422_UNPROCESSABLE_CONTENT,
+    AuthenticationError: status.HTTP_401_UNAUTHORIZED,
+    PermissionDeniedError: status.HTTP_403_FORBIDDEN,
+    DependencyUnavailableError: status.HTTP_503_SERVICE_UNAVAILABLE,
+}
+
+
+def _error_body(code: str, message: str) -> dict[str, dict[str, str]]:
+    return {"error": {"code": code, "message": message}}
+
+
+def register_exception_handlers(app: FastAPI) -> None:
+    @app.exception_handler(DomainError)
+    async def _handle_domain_error(_request: Request, exc: DomainError) -> JSONResponse:
+        http_status = _STATUS_BY_ERROR.get(type(exc), status.HTTP_400_BAD_REQUEST)
+        return JSONResponse(
+            status_code=http_status,
+            content=_error_body(exc.code, exc.message),
+        )
+
+    @app.exception_handler(Exception)
+    async def _handle_unexpected(_request: Request, exc: Exception) -> JSONResponse:
+        log.error("unhandled_exception", exc_info=exc)
+        return JSONResponse(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            content=_error_body("internal_error", "An unexpected error occurred."),
+        )