253 lines
9.0 KiB
Python
253 lines
9.0 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
|
|
"""
|
|
|
|
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"<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)
|
|
|
|
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) |