# 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 - Context-aware HTML error pages """ import logging from fastapi import HTTPException, Request from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse, RedirectResponse from middleware.context import RequestContext, get_request_context from .base import WizamartException from .error_renderer import ErrorPageRenderer logger = logging.getLogger(__name__) def setup_exception_handlers(app): """Setup exception handlers for the FastAPI app.""" @app.exception_handler(WizamartException) async def custom_exception_handler(request: Request, exc: WizamartException): """Handle custom exceptions with context-aware rendering.""" # 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 based on context return _redirect_to_login(request) # 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, "exception_type": type(exc).__name__, }, ) # Check if this is an API request if _is_api_request(request): return JSONResponse(status_code=exc.status_code, content=exc.to_dict()) # Check if this is an HTML page request if _is_html_page_request(request): return ErrorPageRenderer.render_error_page( request=request, status_code=exc.status_code, error_code=exc.error_code, message=exc.message, details=exc.details, show_debug=True, # Will be filtered by ErrorPageRenderer based on user role ) # Default to JSON for unknown request types 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", }, ) # Check if this is an API request if _is_api_request(request): return JSONResponse( status_code=exc.status_code, content={ "error_code": f"HTTP_{exc.status_code}", "message": exc.detail, "status_code": exc.status_code, }, ) # Check if this is an HTML page request if _is_html_page_request(request): return ErrorPageRenderer.render_error_page( request=request, status_code=exc.status_code, error_code=f"HTTP_{exc.status_code}", message=exc.detail, details={}, show_debug=True, ) # Default to JSON 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.""" # Sanitize errors to remove sensitive data from logs sanitized_errors = [] for error in exc.errors(): sanitized_error = error.copy() # Remove 'input' field which may contain passwords if "input" in sanitized_error: sanitized_error["input"] = "" sanitized_errors.append(sanitized_error) logger.error( f"Validation error in {request.method} {request.url}: {len(sanitized_errors)} validation error(s)", extra={ "validation_errors": sanitized_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) # Check if this is an API request if _is_api_request(request): return JSONResponse( status_code=422, content={ "error_code": "VALIDATION_ERROR", "message": "Request validation failed", "status_code": 422, "details": {"validation_errors": clean_errors}, }, ) # Check if this is an HTML page request if _is_html_page_request(request): return ErrorPageRenderer.render_error_page( request=request, status_code=422, error_code="VALIDATION_ERROR", message="Request validation failed", details={"validation_errors": clean_errors}, show_debug=True, ) # Default to JSON return JSONResponse( status_code=422, content={ "error_code": "VALIDATION_ERROR", "message": "Request validation failed", "status_code": 422, "details": {"validation_errors": clean_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__, }, ) # Check if this is an API request if _is_api_request(request): return JSONResponse( status_code=500, content={ "error_code": "INTERNAL_SERVER_ERROR", "message": "Internal server error", "status_code": 500, }, ) # Check if this is an HTML page request if _is_html_page_request(request): return ErrorPageRenderer.render_error_page( request=request, status_code=500, error_code="INTERNAL_SERVER_ERROR", message="Internal server error", details={}, show_debug=True, ) # Default to JSON 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}") # Check if this is an API request if _is_api_request(request): 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}, }, ) # Check if this is an HTML page request if _is_html_page_request(request): return ErrorPageRenderer.render_error_page( request=request, status_code=404, error_code="ENDPOINT_NOT_FOUND", message="The page you're looking for doesn't exist or has been moved.", details={"path": request.url.path}, show_debug=True, ) # Default to JSON 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_api_request(request: Request) -> bool: """ Check if the request is for an API endpoint. API requests ALWAYS return JSON. """ return request.url.path.startswith("/api/") 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 _is_api_request(request): 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") return True def _redirect_to_login(request: Request) -> RedirectResponse: """ Redirect to appropriate login page based on request context. Uses context detection to determine admin vs vendor vs shop login. Properly handles multi-access routing (domain, subdomain, path-based). """ context_type = get_request_context(request) if context_type == RequestContext.ADMIN: logger.debug("Redirecting to /admin/login") return RedirectResponse(url="/admin/login", status_code=302) if context_type == RequestContext.VENDOR_DASHBOARD: logger.debug("Redirecting to /vendor/login") return RedirectResponse(url="/vendor/login", status_code=302) if context_type == RequestContext.SHOP: # For shop context, redirect to shop login (customer login) # Calculate base_url for proper routing (supports domain, subdomain, and path-based access) vendor = getattr(request.state, "vendor", None) vendor_context = getattr(request.state, "vendor_context", None) access_method = ( vendor_context.get("detection_method", "unknown") if vendor_context else "unknown" ) base_url = "/" if access_method == "path" and vendor: full_prefix = ( vendor_context.get("full_prefix", "/vendor/") if vendor_context else "/vendor/" ) base_url = f"{full_prefix}{vendor.subdomain}/" login_url = f"{base_url}shop/account/login" logger.debug(f"Redirecting to {login_url}") return RedirectResponse(url=login_url, status_code=302) # Fallback to root for unknown contexts logger.debug("Unknown context, redirecting to /") return RedirectResponse(url="/", status_code=302) # 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)