84 lines
3.0 KiB
Python
84 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from http import HTTPStatus
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.exceptions import HTTPException
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class AppError(Exception):
|
|
code: str
|
|
message: str
|
|
status_code: int = HTTPStatus.BAD_REQUEST
|
|
|
|
|
|
def _request_id(request: Request) -> str:
|
|
return str(getattr(request.state, "request_id", "unavailable"))
|
|
|
|
|
|
def _response(request: Request, status_code: int, code: str, message: str) -> JSONResponse:
|
|
request_id = _request_id(request)
|
|
return JSONResponse(
|
|
status_code=status_code,
|
|
headers={"X-Request-ID": request_id},
|
|
content={
|
|
"error": {
|
|
"code": code,
|
|
"message": message,
|
|
"request_id": request_id,
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
def install_error_handlers(application: FastAPI) -> None:
|
|
@application.exception_handler(AppError)
|
|
async def application_error(request: Request, error: AppError) -> JSONResponse:
|
|
return _response(request, error.status_code, error.code, error.message)
|
|
|
|
@application.exception_handler(RequestValidationError)
|
|
async def validation_error(request: Request, _error: RequestValidationError) -> JSONResponse:
|
|
return _response(
|
|
request,
|
|
HTTPStatus.UNPROCESSABLE_ENTITY,
|
|
"invalid_request",
|
|
"请求内容不符合要求，请检查后重试。",
|
|
)
|
|
|
|
@application.exception_handler(HTTPException)
|
|
async def http_error(request: Request, error: HTTPException) -> JSONResponse:
|
|
code = {
|
|
HTTPStatus.UNAUTHORIZED: "authentication_required",
|
|
HTTPStatus.FORBIDDEN: "access_denied",
|
|
HTTPStatus.NOT_FOUND: "not_found",
|
|
HTTPStatus.CONFLICT: "conflict",
|
|
}.get(error.status_code, "request_failed")
|
|
message = {
|
|
HTTPStatus.UNAUTHORIZED: "请先登录。",
|
|
HTTPStatus.FORBIDDEN: "当前账号无权执行此操作。",
|
|
HTTPStatus.NOT_FOUND: "请求的内容不存在。",
|
|
HTTPStatus.CONFLICT: "当前状态已发生变化，请刷新后重试。",
|
|
}.get(error.status_code, "请求失败，请稍后重试。")
|
|
return _response(request, error.status_code, code, message)
|
|
|
|
@application.exception_handler(Exception)
|
|
async def unexpected_error(request: Request, error: Exception) -> JSONResponse:
|
|
logger.exception(
|
|
"Unhandled request failure",
|
|
exc_info=(type(error), error, error.__traceback__),
|
|
extra={"request_id": _request_id(request), "event": "request.failed"},
|
|
)
|
|
return _response(
|
|
request,
|
|
HTTPStatus.INTERNAL_SERVER_ERROR,
|
|
"internal_error",
|
|
"服务暂时不可用，请稍后重试。",
|
|
)
|