Samir Boulahtit
b769f5a047
Major architecture change to unify frontend detection: ## Problem Solved - Eliminated code duplication across 3 middleware files - Fixed incomplete path detection (now detects /api/v1/admin/*) - Unified on FrontendType enum (deprecates RequestContext) - Added request.state.frontend_type for all requests ## New Components - app/core/frontend_detector.py: Centralized FrontendDetector class - middleware/frontend_type.py: FrontendTypeMiddleware (replaces ContextMiddleware) - docs/architecture/frontend-detection.md: Complete architecture documentation ## Changes - main.py: Use FrontendTypeMiddleware instead of ContextMiddleware - middleware/context.py: Deprecated (kept for backwards compatibility) - middleware/platform_context.py: Use FrontendDetector.is_admin() - middleware/vendor_context.py: Use FrontendDetector.is_admin() - middleware/language.py: Use FrontendType instead of context_value - app/exceptions/handler.py: Use FrontendType.STOREFRONT - app/exceptions/error_renderer.py: Use FrontendType - Customer routes: Cookie path changed from /shop to /storefront ## Documentation - docs/architecture/frontend-detection.md: New comprehensive docs - docs/architecture/middleware.md: Updated for new system - docs/architecture/request-flow.md: Updated for FrontendType - docs/backend/middleware-reference.md: Updated API reference ## Tests - tests/unit/core/test_frontend_detector.py: 37 new tests - tests/unit/middleware/test_frontend_type.py: 11 new tests - tests/unit/middleware/test_context.py: Updated for compatibility Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
478 lines
17 KiB
Python
478 lines
17 KiB
Python
# 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 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 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., vendor 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",
|
|
# Vendor-level auth errors
|
|
"VENDOR_ACCESS_DENIED",
|
|
"UNAUTHORIZED_VENDOR_ACCESS",
|
|
"VENDOR_OWNER_ONLY",
|
|
"INSUFFICIENT_VENDOR_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"] = "<redacted>"
|
|
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"<bytes: {len(value)} bytes>"
|
|
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"<bytes: {len(value)} bytes>"
|
|
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 vendor vs storefront login.
|
|
Properly handles multi-access routing (domain, subdomain, path-based).
|
|
"""
|
|
frontend_type = get_frontend_type(request)
|
|
|
|
if frontend_type == FrontendType.ADMIN:
|
|
logger.debug("Redirecting to /admin/login")
|
|
return RedirectResponse(url="/admin/login", status_code=302)
|
|
if frontend_type == FrontendType.VENDOR:
|
|
# Extract vendor code from the request path
|
|
# Path format: /vendor/{vendor_code}/...
|
|
path_parts = request.url.path.split("/")
|
|
vendor_code = None
|
|
|
|
# Find vendor code in path
|
|
if len(path_parts) >= 3 and path_parts[1] == "vendor":
|
|
vendor_code = path_parts[2]
|
|
|
|
# Fallback: try to get from request state
|
|
if not vendor_code:
|
|
vendor = getattr(request.state, "vendor", None)
|
|
if vendor:
|
|
vendor_code = vendor.subdomain
|
|
|
|
# Construct proper login URL with vendor code
|
|
if vendor_code:
|
|
login_url = f"/vendor/{vendor_code}/login"
|
|
else:
|
|
# Fallback if we can't determine vendor code
|
|
login_url = "/vendor/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)
|
|
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}storefront/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)
|