"""Custom application exceptions and FastAPI exception handlers.""" from __future__ import annotations from fastapi import FastAPI, Request, status from fastapi.responses import JSONResponse from app.core.logging import get_logger logger = get_logger(__name__) class AppException(Exception): """Base class for all expected, handled application errors.""" status_code: int = status.HTTP_400_BAD_REQUEST detail: str = "Application error" def __init__(self, detail: str | None = None) -> None: if detail is not None: self.detail = detail super().__init__(self.detail) class NotFoundError(AppException): status_code = status.HTTP_404_NOT_FOUND detail = "Resource not found" class ConflictError(AppException): status_code = status.HTTP_409_CONFLICT detail = "Resource already exists" class AuthenticationError(AppException): status_code = status.HTTP_401_UNAUTHORIZED detail = "Could not validate credentials" class AuthorizationError(AppException): status_code = status.HTTP_403_FORBIDDEN detail = "Not enough permissions" class ValidationError(AppException): status_code = status.HTTP_422_UNPROCESSABLE_ENTITY detail = "Validation failed" class ServiceUnavailableError(AppException): status_code = status.HTTP_503_SERVICE_UNAVAILABLE detail = "Service temporarily unavailable" def register_exception_handlers(app: FastAPI) -> None: @app.exception_handler(AppException) async def _app_exception_handler(request: Request, exc: AppException) -> JSONResponse: if exc.status_code >= 500: logger.error("app_exception", detail=exc.detail, path=request.url.path) return JSONResponse( status_code=exc.status_code, content={"detail": exc.detail}, ) @app.exception_handler(Exception) async def _unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse: logger.exception("unhandled_exception", path=request.url.path) return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"detail": "Internal server error"}, )