| """
|
| Global error handlers and exceptions
|
| """
|
| from fastapi import FastAPI, Request, status
|
| from fastapi.responses import JSONResponse
|
| from sqlalchemy.exc import SQLAlchemyError
|
| from typing import Union, Dict, Any
|
|
|
| class AppException(Exception):
|
| """Base application exception"""
|
| def __init__(
|
| self,
|
| status_code: int,
|
| detail: Union[str, Dict[str, Any]],
|
| headers: Dict[str, str] = None
|
| ):
|
| self.status_code = status_code
|
| self.detail = detail
|
| self.headers = headers
|
|
|
| def setup_error_handlers(app: FastAPI):
|
| """Setup global error handlers"""
|
|
|
| @app.exception_handler(AppException)
|
| async def app_exception_handler(request: Request, exc: AppException):
|
| headers = exc.headers if exc.headers else {}
|
| return JSONResponse(
|
| status_code=exc.status_code,
|
| content={"detail": exc.detail},
|
| headers=headers
|
| )
|
|
|
| @app.exception_handler(SQLAlchemyError)
|
| async def sqlalchemy_exception_handler(request: Request, exc: SQLAlchemyError):
|
| return JSONResponse(
|
| status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
| content={
|
| "detail": "Database error occurred",
|
| "message": str(exc)
|
| }
|
| )
|
|
|
| @app.exception_handler(Exception)
|
| async def general_exception_handler(request: Request, exc: Exception):
|
| return JSONResponse(
|
| status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
| content={
|
| "detail": "An unexpected error occurred",
|
| "message": str(exc)
|
| }
|
| )
|
|
|
