# 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 app.modules.enums import FrontendType from middleware.frontend_type import get_frontend_type from .base import OrionException from .error_renderer import ErrorPageRenderer logger = logging.getLogger(__name__) def setup_exception_handlers(app): """Setup exception handlers for the FastAPI app.""" @app.exception_handler(OrionException) async def custom_exception_handler(request: Request, exc: OrionException): """Handle custom exceptions with context-aware rendering.""" # Special handling for auth errors on HTML page requests (redirect to login) # This includes both: # - 401 errors: Not authenticated (expired/invalid token) # - 403 errors with specific auth codes: Authenticated but wrong context # (e.g., store token on admin page, role mismatch) # These codes indicate the user should re-authenticate with correct credentials auth_redirect_error_codes = { # Auth-level errors "ADMIN_REQUIRED", "INSUFFICIENT_PERMISSIONS", "USER_NOT_ACTIVE", # Store-level auth errors "STORE_ACCESS_DENIED", "UNAUTHORIZED_STORE_ACCESS", "STORE_OWNER_ONLY", "INSUFFICIENT_STORE_PERMISSIONS", # Customer-level auth errors "CUSTOMER_NOT_AUTHORIZED", } should_redirect = ( _is_html_page_request(request) and ( exc.status_code == 401 or (exc.status_code == 403 and exc.error_code in auth_redirect_error_codes) ) ) if should_redirect: logger.info( f"Auth error on HTML page request - redirecting to login: {request.url.path}", extra={ "path": request.url.path, "status_code": exc.status_code, "error_code": exc.error_code, "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 frontend type. Uses FrontendType detection to determine admin vs store vs storefront login. Properly handles multi-access routing (domain, subdomain, path-based). """ frontend_type = get_frontend_type(request) or FrontendType.PLATFORM if frontend_type == FrontendType.ADMIN: logger.debug("Redirecting to /admin/login") return RedirectResponse(url="/admin/login", status_code=302) if frontend_type == FrontendType.MERCHANT: logger.debug("Redirecting to /merchants/login") return RedirectResponse(url="/merchants/login", status_code=302) if frontend_type == FrontendType.STORE: # Extract store code from the request path # Path format: /store/{store_code}/... path_parts = request.url.path.split("/") store_code = None # Find store code in path if len(path_parts) >= 3 and path_parts[1] == "store": store_code = path_parts[2] # Fallback: try to get from request state if not store_code: store = getattr(request.state, "store", None) if store: store_code = store.subdomain # Construct proper login URL with store code if store_code: login_url = f"/store/{store_code}/login" else: # Fallback if we can't determine store code login_url = "/store/login" logger.debug(f"Redirecting to {login_url}") return RedirectResponse(url=login_url, status_code=302) if frontend_type == FrontendType.STOREFRONT: # For storefront context, redirect to storefront login (customer login) # Calculate base_url for proper routing (supports domain, subdomain, and path-based access) store = getattr(request.state, "store", None) store_context = getattr(request.state, "store_context", None) access_method = ( store_context.get("detection_method", "unknown") if store_context else "unknown" ) base_url = "/" if access_method == "path" and store: platform = getattr(request.state, "platform", None) platform_original_path = getattr(request.state, "platform_original_path", None) if platform and platform_original_path and platform_original_path.startswith("/platforms/"): base_url = f"/platforms/{platform.code}/storefront/{store.store_code}/" else: full_prefix = ( store_context.get("full_prefix", "/storefront/") if store_context else "/storefront/" ) base_url = f"{full_prefix}{store.store_code}/" login_url = f"{base_url}account/login" logger.debug(f"Redirecting to {login_url}") return RedirectResponse(url=login_url, status_code=302) # Fallback to root for unknown contexts (PLATFORM) logger.debug("Platform 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)