Senko-san
62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
"""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,
|
|
RangeNotSatisfiableError,
|
|
StorageError,
|
|
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,
|
|
StorageError: status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
}
|
|
|
|
|
|
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(RangeNotSatisfiableError)
|
|
async def _handle_range_error(_request: Request, exc: RangeNotSatisfiableError) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=status.HTTP_416_REQUESTED_RANGE_NOT_SATISFIABLE,
|
|
content=_error_body(exc.code, exc.message),
|
|
headers={"Content-Range": f"bytes */{exc.total_size}"},
|
|
)
|
|
|
|
@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."),
|
|
)
|