Add custom exceptions handling
This commit is contained in:
207
app/exceptions/handler.py
Normal file
207
app/exceptions/handler.py
Normal file
@@ -0,0 +1,207 @@
|
||||
# app/exceptions/handler.py
|
||||
"""
|
||||
Exception handler for FastAPI application.
|
||||
|
||||
Provides consistent error responses and logging for all custom exceptions.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Union
|
||||
|
||||
from fastapi import Request, HTTPException
|
||||
from fastapi.responses import JSONResponse
|
||||
from starlette.middleware.base import BaseHTTPMiddleware
|
||||
|
||||
from .base import LetzShopException
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ExceptionHandlerMiddleware(BaseHTTPMiddleware):
|
||||
"""Middleware to handle custom exceptions and convert them to JSON responses."""
|
||||
|
||||
async def dispatch(self, request: Request, call_next):
|
||||
try:
|
||||
response = await call_next(request)
|
||||
return response
|
||||
except LetzShopException as exc:
|
||||
return await self.handle_custom_exception(request, exc)
|
||||
except HTTPException as exc:
|
||||
return await self.handle_http_exception(request, exc)
|
||||
except Exception as exc:
|
||||
return await self.handle_generic_exception(request, exc)
|
||||
|
||||
async def handle_custom_exception(
|
||||
self,
|
||||
request: Request,
|
||||
exc: LetzShopException
|
||||
) -> JSONResponse:
|
||||
"""Handle custom LetzShop exceptions."""
|
||||
|
||||
# Log the exception
|
||||
logger.error(
|
||||
f"Custom exception in {request.method} {request.url}: "
|
||||
f"{exc.error_code} - {exc.message}",
|
||||
extra={
|
||||
"error_code": exc.error_code,
|
||||
"status_code": exc.status_code,
|
||||
"details": exc.details,
|
||||
"url": str(request.url),
|
||||
"method": request.method,
|
||||
}
|
||||
)
|
||||
|
||||
return JSONResponse(
|
||||
status_code=exc.status_code,
|
||||
content=exc.to_dict()
|
||||
)
|
||||
|
||||
async def handle_http_exception(
|
||||
self,
|
||||
request: Request,
|
||||
exc: HTTPException
|
||||
) -> JSONResponse:
|
||||
"""Handle FastAPI HTTPExceptions."""
|
||||
|
||||
logger.error(
|
||||
f"HTTP exception in {request.method} {request.url}: "
|
||||
f"{exc.status_code} - {exc.detail}",
|
||||
extra={
|
||||
"status_code": exc.status_code,
|
||||
"detail": exc.detail,
|
||||
"url": str(request.url),
|
||||
"method": request.method,
|
||||
}
|
||||
)
|
||||
|
||||
return JSONResponse(
|
||||
status_code=exc.status_code,
|
||||
content={
|
||||
"error_code": f"HTTP_{exc.status_code}",
|
||||
"message": exc.detail,
|
||||
"status_code": exc.status_code,
|
||||
}
|
||||
)
|
||||
|
||||
async def handle_generic_exception(
|
||||
self,
|
||||
request: Request,
|
||||
exc: Exception
|
||||
) -> JSONResponse:
|
||||
"""Handle unexpected exceptions."""
|
||||
|
||||
logger.error(
|
||||
f"Unexpected exception in {request.method} {request.url}: {str(exc)}",
|
||||
exc_info=True,
|
||||
extra={
|
||||
"url": str(request.url),
|
||||
"method": request.method,
|
||||
"exception_type": type(exc).__name__,
|
||||
}
|
||||
)
|
||||
|
||||
return JSONResponse(
|
||||
status_code=500,
|
||||
content={
|
||||
"error_code": "INTERNAL_SERVER_ERROR",
|
||||
"message": "Internal server error",
|
||||
"status_code": 500,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def setup_exception_handlers(app):
|
||||
"""Setup exception handlers for the FastAPI app."""
|
||||
|
||||
@app.exception_handler(LetzShopException)
|
||||
async def custom_exception_handler(request: Request, exc: LetzShopException):
|
||||
"""Handle custom LetzShop exceptions."""
|
||||
|
||||
logger.error(
|
||||
f"Custom exception in {request.method} {request.url}: "
|
||||
f"{exc.error_code} - {exc.message}",
|
||||
extra={
|
||||
"error_code": exc.error_code,
|
||||
"status_code": exc.status_code,
|
||||
"details": exc.details,
|
||||
"url": str(request.url),
|
||||
"method": request.method,
|
||||
}
|
||||
)
|
||||
|
||||
return JSONResponse(
|
||||
status_code=exc.status_code,
|
||||
content=exc.to_dict()
|
||||
)
|
||||
|
||||
@app.exception_handler(HTTPException)
|
||||
async def http_exception_handler(request: Request, exc: HTTPException):
|
||||
"""Handle FastAPI HTTPExceptions with consistent format."""
|
||||
|
||||
logger.error(
|
||||
f"HTTP exception in {request.method} {request.url}: "
|
||||
f"{exc.status_code} - {exc.detail}",
|
||||
extra={
|
||||
"status_code": exc.status_code,
|
||||
"detail": exc.detail,
|
||||
"url": str(request.url),
|
||||
"method": request.method,
|
||||
}
|
||||
)
|
||||
|
||||
return JSONResponse(
|
||||
status_code=exc.status_code,
|
||||
content={
|
||||
"error_code": f"HTTP_{exc.status_code}",
|
||||
"message": exc.detail,
|
||||
"status_code": exc.status_code,
|
||||
}
|
||||
)
|
||||
|
||||
@app.exception_handler(Exception)
|
||||
async def generic_exception_handler(request: Request, exc: Exception):
|
||||
"""Handle unexpected exceptions."""
|
||||
|
||||
logger.error(
|
||||
f"Unexpected exception in {request.method} {request.url}: {str(exc)}",
|
||||
exc_info=True,
|
||||
extra={
|
||||
"url": str(request.url),
|
||||
"method": request.method,
|
||||
"exception_type": type(exc).__name__,
|
||||
}
|
||||
)
|
||||
|
||||
return JSONResponse(
|
||||
status_code=500,
|
||||
content={
|
||||
"error_code": "INTERNAL_SERVER_ERROR",
|
||||
"message": "Internal server error",
|
||||
"status_code": 500,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
# Utility functions for common exception scenarios
|
||||
def raise_not_found(resource_type: str, identifier: str) -> None:
|
||||
"""Convenience function to raise ResourceNotFoundException."""
|
||||
from .base import ResourceNotFoundException
|
||||
raise ResourceNotFoundException(resource_type, identifier)
|
||||
|
||||
|
||||
def raise_validation_error(message: str, field: str = None, details: dict = None) -> None:
|
||||
"""Convenience function to raise ValidationException."""
|
||||
from .base import ValidationException
|
||||
raise ValidationException(message, field, details)
|
||||
|
||||
|
||||
def raise_auth_error(message: str = "Authentication failed") -> None:
|
||||
"""Convenience function to raise AuthenticationException."""
|
||||
from .base import AuthenticationException
|
||||
raise AuthenticationException(message)
|
||||
|
||||
|
||||
def raise_permission_error(message: str = "Access denied") -> None:
|
||||
"""Convenience function to raise AuthorizationException."""
|
||||
from .base import AuthorizationException
|
||||
raise AuthorizationException(message)
|
||||
Reference in New Issue
Block a user