# app/exceptions/handler.py """ Exception handler for FastAPI application. Provides consistent error responses and logging for all custom exceptions. This module provides classes and functions for: - Unified exception handling for all application exceptions - Consistent error response formatting - Comprehensive logging with structured data """ import logging from typing import Union from fastapi import Request, HTTPException from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse, RedirectResponse from .base import LetzShopException logger = logging.getLogger(__name__) 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 exceptions.""" # Special handling for 401 on HTML page requests (redirect to login) if exc.status_code == 401 and _is_html_page_request(request): logger.info( f"401 on HTML page request - redirecting to login: {request.url.path}", extra={ "path": request.url.path, "accept": request.headers.get("accept", ""), "method": request.method } ) # Redirect to appropriate login page if request.url.path.startswith("/admin"): logger.debug("Redirecting to /admin/login") return RedirectResponse(url="/admin/login", status_code=302) elif "/vendor/" in request.url.path: logger.debug("Redirecting to /vendor/login") return RedirectResponse(url="/vendor/login", status_code=302) # If neither, fall through to JSON response logger.debug("No specific redirect path matched, returning JSON") 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, "exception_type": type(exc).__name__, } ) 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, "exception_type": "HTTPException", } ) 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(RequestValidationError) async def validation_exception_handler(request: Request, exc: RequestValidationError): """Handle Pydantic validation errors with consistent format.""" logger.error( f"Validation error in {request.method} {request.url}: {exc.errors()}", extra={ "validation_errors": exc.errors(), "url": str(request.url), "method": request.method, "exception_type": "RequestValidationError", } ) # Clean up validation errors to ensure JSON serializability clean_errors = [] for error in exc.errors(): clean_error = {} for key, value in error.items(): if key == 'input' and isinstance(value, bytes): # Convert bytes to string representation for JSON serialization clean_error[key] = f"" elif key == 'ctx' and isinstance(value, dict): # Handle the 'ctx' field that contains ValueError objects clean_ctx = {} for ctx_key, ctx_value in value.items(): if isinstance(ctx_value, Exception): clean_ctx[ctx_key] = str(ctx_value) # Convert exception to string else: clean_ctx[ctx_key] = ctx_value clean_error[key] = clean_ctx elif isinstance(value, bytes): # Handle any other bytes objects clean_error[key] = f"" else: clean_error[key] = value clean_errors.append(clean_error) return JSONResponse( status_code=422, content={ "error_code": "VALIDATION_ERROR", "message": "Request validation failed", "status_code": 422, "details": { "validation_errors": clean_errors # Use cleaned errors } } ) @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, } ) @app.exception_handler(404) async def not_found_handler(request: Request, exc): """Handle all 404 errors with consistent format.""" logger.warning(f"404 Not Found: {request.method} {request.url}") return JSONResponse( status_code=404, content={ "error_code": "ENDPOINT_NOT_FOUND", "message": f"Endpoint not found: {request.url.path}", "status_code": 404, "details": { "path": request.url.path, "method": request.method } } ) def _is_html_page_request(request: Request) -> bool: """ Check if the request is for an HTML page (not an API endpoint). More precise detection: - Must NOT have /api/ in path - Must be GET request - Must explicitly accept text/html - Must not already be on login page """ logger.debug( f"Checking if HTML page request: {request.url.path}", extra={ "path": request.url.path, "method": request.method, "accept": request.headers.get("accept", "") } ) # Don't redirect API calls if "/api/" in request.url.path: logger.debug("Not HTML page: API endpoint") return False # Don't redirect if already on login page if request.url.path.endswith("/login"): logger.debug("Not HTML page: Already on login page") return False # Only redirect GET requests (page loads) if request.method != "GET": logger.debug(f"Not HTML page: Method is {request.method}, not GET") return False # MUST explicitly accept HTML (strict check) accept_header = request.headers.get("accept", "") if "text/html" not in accept_header: logger.debug(f"Not HTML page: Accept header doesn't include text/html: {accept_header}") return False logger.debug("IS HTML page request - will redirect on 401") return True # 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)