import logging from http import HTTPStatus from typing import cast from fastapi import FastAPI from fastapi.exceptions import RequestValidationError from fastapi.utils import is_body_allowed_for_status_code from pydantic_core import ErrorDetails from starlette import status from starlette.exceptions import HTTPException from starlette.requests import Request from starlette.responses import JSONResponse, Response from fansifter_common.exceptions import FansifterError, InvalidInputError logger = logging.getLogger(__name__) async def default_error_handler(request: Request, exc: Exception) -> Response: return JSONResponse( content={ "code": "internal_error", "message": ( "The server encountered an internal error " "and was unable to complete your request." ), }, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, ) async def http_exception_handler(request: Request, exc: HTTPException) -> Response: if not is_body_allowed_for_status_code(exc.status_code): return Response(status_code=exc.status_code, headers=exc.headers) return JSONResponse( { "code": HTTPStatus(exc.status_code).name.lower(), "message": exc.detail, }, status_code=exc.status_code, headers=exc.headers, ) async def fansifter_error_handler(request: Request, exc: FansifterError) -> Response: if exc.status_code >= status.HTTP_500_INTERNAL_SERVER_ERROR: logger.error(exc.message, extra={"code": exc.code}, exc_info=exc) return JSONResponse(content=exc.dump(), status_code=exc.status_code) async def request_validation_error_handler( request: Request, exc: RequestValidationError ) -> Response: return await fansifter_error_handler( request, exc=InvalidInputError.from_pydantic( cast(list[ErrorDetails], exc.errors()), loc_pos=1, status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, ), ) def register_defaults(app: FastAPI) -> None: app.add_exception_handler(Exception, default_error_handler) app.add_exception_handler(HTTPException, http_exception_handler) # type: ignore[arg-type] # ty: ignore[invalid-argument-type] # noqa app.add_exception_handler(RequestValidationError, request_validation_error_handler) # type: ignore[arg-type] # ty: ignore[invalid-argument-type] # noqa app.add_exception_handler(FansifterError, fansifter_error_handler) # type: ignore[arg-type] # ty: ignore[invalid-argument-type] # noqa