104 lines
2.6 KiB
Python
104 lines
2.6 KiB
Python
from http import HTTPStatus
|
|
import re
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from app.api.router import api_router
|
|
from app.core.config import settings
|
|
from app.schemas.errors import ApiErrorResponse
|
|
|
|
|
|
def _error_name(status_code: int) -> str:
|
|
try:
|
|
phrase = HTTPStatus(status_code).phrase
|
|
except ValueError:
|
|
return "HTTP_ERROR"
|
|
return re.sub(r"[^A-Z0-9]+", "_", phrase.upper()).strip("_")
|
|
|
|
|
|
def _error_message(detail: Any) -> str | list[str]:
|
|
if isinstance(detail, str):
|
|
return detail
|
|
if isinstance(detail, list):
|
|
messages = [
|
|
str(item.get("msg", item)) if isinstance(item, dict) else str(item)
|
|
for item in detail
|
|
]
|
|
return messages or "Request validation failed."
|
|
return str(detail)
|
|
|
|
|
|
def _error_payload(
|
|
request: Request,
|
|
status_code: int,
|
|
message: str | list[str],
|
|
error: str,
|
|
) -> dict[str, Any]:
|
|
return ApiErrorResponse(
|
|
code=error,
|
|
message=message,
|
|
error=error,
|
|
statusCode=status_code,
|
|
path=request.url.path,
|
|
).model_dump(by_alias=True)
|
|
|
|
|
|
async def _http_exception_handler(
|
|
request: Request,
|
|
exception: HTTPException,
|
|
) -> JSONResponse:
|
|
error = _error_name(exception.status_code)
|
|
return JSONResponse(
|
|
status_code=exception.status_code,
|
|
content=_error_payload(
|
|
request,
|
|
exception.status_code,
|
|
_error_message(exception.detail),
|
|
error,
|
|
),
|
|
)
|
|
|
|
|
|
async def _validation_exception_handler(
|
|
request: Request,
|
|
exception: RequestValidationError,
|
|
) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=422,
|
|
content=_error_payload(
|
|
request,
|
|
422,
|
|
_error_message(exception.errors()),
|
|
"VALIDATION_ERROR",
|
|
),
|
|
)
|
|
|
|
|
|
def _register_exception_handlers(application: FastAPI) -> None:
|
|
application.add_exception_handler(HTTPException, _http_exception_handler)
|
|
application.add_exception_handler(
|
|
RequestValidationError,
|
|
_validation_exception_handler,
|
|
)
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
error_responses = {
|
|
status_code: {"model": ApiErrorResponse}
|
|
for status_code in (400, 401, 403, 404, 422, 500)
|
|
}
|
|
application = FastAPI(
|
|
title=settings.app_name,
|
|
root_path=settings.docs_root_path,
|
|
responses=error_responses,
|
|
)
|
|
_register_exception_handlers(application)
|
|
application.include_router(api_router)
|
|
return application
|
|
|
|
|
|
app = create_app()
|