| """ |
| Global exception handlers and error formatting. |
| """ |
| import logging |
| import traceback |
|
|
| from fastapi import FastAPI, Request, status |
| from fastapi.exceptions import RequestValidationError |
| from fastapi.responses import JSONResponse |
| from starlette.exceptions import HTTPException as StarletteHTTPException |
|
|
| from app.core.config import settings |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| def register_exception_handlers(app: FastAPI): |
| """Register all global exception handlers.""" |
|
|
| @app.exception_handler(StarletteHTTPException) |
| async def http_exception_handler(request: Request, exc: StarletteHTTPException): |
| """Handle HTTP exceptions with consistent format.""" |
| return JSONResponse( |
| status_code=exc.status_code, |
| content={ |
| "error": True, |
| "status_code": exc.status_code, |
| "detail": exc.detail, |
| "path": str(request.url.path), |
| }, |
| headers=getattr(exc, "headers", None), |
| ) |
|
|
| @app.exception_handler(RequestValidationError) |
| async def validation_exception_handler(request: Request, exc: RequestValidationError): |
| """Handle Pydantic validation errors with readable messages.""" |
| errors = [] |
| for error in exc.errors(): |
| field = " → ".join(str(loc) for loc in error["loc"][1:]) |
| errors.append({ |
| "field": field, |
| "message": error["msg"], |
| "type": error["type"], |
| }) |
|
|
| return JSONResponse( |
| status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, |
| content={ |
| "error": True, |
| "status_code": 422, |
| "detail": "Validation error", |
| "errors": errors, |
| "path": str(request.url.path), |
| }, |
| ) |
|
|
| @app.exception_handler(Exception) |
| async def general_exception_handler(request: Request, exc: Exception): |
| """Handle unexpected exceptions.""" |
| error_id = id(exc) |
| logger.error( |
| f"Unhandled exception [{error_id}]: {exc}\n" |
| f"Path: {request.url.path}\n" |
| f"{''.join(traceback.format_exception(type(exc), exc, exc.__traceback__))}" |
| ) |
|
|
| |
| if settings.APP_ENV == "production": |
| detail = "An unexpected error occurred. Please try again later." |
| else: |
| detail = f"{type(exc).__name__}: {str(exc)}" |
|
|
| |
| try: |
| import sentry_sdk |
| sentry_sdk.capture_exception(exc) |
| except ImportError: |
| pass |
|
|
| return JSONResponse( |
| status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, |
| content={ |
| "error": True, |
| "status_code": 500, |
| "detail": detail, |
| "error_id": str(error_id), |
| "path": str(request.url.path), |
| }, |
| ) |
|
|
