File size: 1,815 Bytes
f3fa932 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | """
Custom exception handlers for the FastAPI application.
"""
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
import logging
logger = logging.getLogger("llm_firewall.errors")
async def http_exception_handler(request: Request, exc: HTTPException):
"""Handle HTTP exceptions with consistent format."""
return JSONResponse(
status_code=exc.status_code,
content={
"error": _status_to_error(exc.status_code),
"detail": exc.detail if isinstance(exc.detail, str) else str(exc.detail),
},
headers=getattr(exc, "headers", None),
)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
"""Handle Pydantic validation errors."""
errors = exc.errors()
detail = "; ".join(
f"{err.get('loc', ['unknown'])[-1]}: {err.get('msg', 'invalid')}"
for err in errors
)
return JSONResponse(
status_code=422,
content={"error": "validation_error", "detail": detail},
)
async def general_exception_handler(request: Request, exc: Exception):
"""Catch-all for unhandled exceptions."""
logger.exception("Unhandled exception:")
return JSONResponse(
status_code=500,
content={
"error": "internal_error",
"detail": str(exc)
})
def _status_to_error(status_code: int) -> str:
"""Map HTTP status code to error string."""
mapping = {
400: "bad_request",
401: "unauthorized",
403: "forbidden",
404: "not_found",
429: "rate_limit_exceeded",
500: "internal_error",
502: "bad_gateway",
503: "service_unavailable",
}
return mapping.get(status_code, "error")
|