import logging from typing import Iterable, Optional, Tuple, cast from fastapi.exception_handlers import ( http_exception_handler as fastapi_http_exception_handler, ) from fastapi.exceptions import RequestValidationError from starlette import status from starlette.exceptions import HTTPException from starlette.requests import Request from service.api.responses import JSONResponse logger = logging.getLogger(__name__) async def validation_exception_handler(request: Request, exc: RequestValidationError): loc = cast(Iterable[str], exc.errors()[0]["loc"]) field_name = ".".join(loc) error = f"{field_name} - {exc.errors()[0]['msg']}" type_, info = _get_exc_type_info(exc) return JSONResponse( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, content={ "error": error, "type": type_, "info": info, }, ) async def value_exception_handler(request: Request, exc: ValueError): type_, info = _get_exc_type_info(exc) return JSONResponse( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, content={ "error": str(exc), "type": type_, "info": info, }, ) async def http_exception_handler(request: Request, exc: HTTPException): type_, info = _get_exc_type_info(exc) if exc.status_code == status.HTTP_401_UNAUTHORIZED: return await fastapi_http_exception_handler(request, exc) return JSONResponse( status_code=exc.status_code, content={ "error": exc.detail, "type": type_, "info": info, }, ) async def exception_handler(request: Request, exc: Exception): """ Common exception handler. This will work only debug is disable. Otherwise, you will get stacktrace. """ logger.exception(f"API common exception: {exc}") type_, info = _get_exc_type_info(exc) return JSONResponse( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, content={ "error": str(exc.args[0]) if len(exc.args) > 0 else None, "type": type_, "info": info, }, ) def _get_exc_type_info(exc: Exception) -> Tuple[str, Optional[str]]: return exc.__class__.__name__, exc.args[1] if len(exc.args) > 1 else None