124 lines
4.6 KiB
Python
124 lines
4.6 KiB
Python
from fastapi import Request, status
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.exceptions import RequestValidationError
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from starlette.exceptions import HTTPException as StarletteHTTPException
|
|
import traceback
|
|
|
|
from exceptions import (
|
|
ApiException,
|
|
BusinessValidationException,
|
|
ValidationException
|
|
)
|
|
from schemas.ApiResponse import ApiResponse
|
|
|
|
|
|
class ErrorHandlerMiddleware(BaseHTTPMiddleware):
|
|
"""Middleware for handling exceptions globally in the application."""
|
|
|
|
async def dispatch(self, request: Request, call_next):
|
|
try:
|
|
return await call_next(request)
|
|
except Exception as exc:
|
|
return self.handle_exception(exc)
|
|
|
|
def handle_exception(self, exc: Exception) -> JSONResponse:
|
|
if isinstance(exc, ApiException):
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content=ApiResponse.from_api_exception(exc)
|
|
)
|
|
elif isinstance(exc, StarletteHTTPException):
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content=ApiResponse(
|
|
message=str(exc.detail),
|
|
error=str(traceback.format_exc())
|
|
).model_dump(exclude_none=True)
|
|
)
|
|
elif isinstance(exc, (ValidationException, BusinessValidationException)):
|
|
return JSONResponse(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
content=ApiResponse(
|
|
message=str(exc),
|
|
error=str(traceback.format_exc())
|
|
).model_dump(exclude_none=True)
|
|
)
|
|
elif isinstance(exc, RequestValidationError):
|
|
return JSONResponse(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
content=ApiResponse(
|
|
message="Validation error",
|
|
error=str(exc.errors())
|
|
).model_dump(exclude_none=True)
|
|
)
|
|
else:
|
|
return JSONResponse(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
content=ApiResponse(
|
|
message=str(exc),
|
|
error=str(traceback.format_exc())
|
|
).model_dump(exclude_none=True)
|
|
)
|
|
|
|
|
|
# Exception handlers for FastAPI
|
|
def configure_exception_handlers(app):
|
|
"""Configure exception handlers for the FastAPI application."""
|
|
|
|
@app.exception_handler(ApiException)
|
|
async def api_exception_handler(request: Request, exc: ApiException):
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content=ApiResponse.from_api_exception(exc)
|
|
)
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def validation_exception_handler(request: Request, exc: RequestValidationError):
|
|
return JSONResponse(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
content=ApiResponse(
|
|
message="Validation error",
|
|
error=str(exc.errors())
|
|
).model_dump(exclude_none=True)
|
|
)
|
|
|
|
@app.exception_handler(StarletteHTTPException)
|
|
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content=ApiResponse(
|
|
message=str(exc.detail),
|
|
error=str(traceback.format_exc())
|
|
).model_dump(exclude_none=True)
|
|
)
|
|
|
|
@app.exception_handler(BusinessValidationException)
|
|
async def business_validation_exception_handler(request: Request, exc: BusinessValidationException):
|
|
return JSONResponse(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
content=ApiResponse(
|
|
message=str(exc),
|
|
error=str(traceback.format_exc())
|
|
).model_dump(exclude_none=True)
|
|
)
|
|
|
|
@app.exception_handler(ValidationException)
|
|
async def custom_validation_exception_handler(request: Request, exc: ValidationException):
|
|
return JSONResponse(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
content=ApiResponse(
|
|
message=str(exc),
|
|
error=str(traceback.format_exc())
|
|
).model_dump(exclude_none=True)
|
|
)
|
|
|
|
@app.exception_handler(Exception)
|
|
async def general_exception_handler(request: Request, exc: Exception):
|
|
return JSONResponse(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
content=ApiResponse(
|
|
message=str(exc),
|
|
error=str(traceback.format_exc())
|
|
).model_dump(exclude_none=True)
|
|
) |