zk
74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
import traceback
|
|
|
|
from fastapi import Request
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.exceptions import HTTPException as StarletteHTTPException
|
|
|
|
from app.config import settings
|
|
from app.core.logger import log
|
|
from app.core.schemas.responses import StandardResponse
|
|
|
|
# 友好的 HTTP 状态码消息映射
|
|
_FRIENDLY_MESSAGES = {
|
|
400: "请求参数错误",
|
|
401: "未经授权，请登录",
|
|
403: "权限不足，禁止访问",
|
|
404: "请求的资源不存在",
|
|
405: "请求方法不允许",
|
|
422: "数据验证失败",
|
|
429: "请求过于频繁",
|
|
500: "服务器内部错误",
|
|
}
|
|
|
|
|
|
def _get_uuid(request: Request) -> str | None:
|
|
return getattr(request.state, "uuid", None)
|
|
|
|
|
|
async def http_exception_handler(request: Request, exc: StarletteHTTPException) -> JSONResponse:
|
|
uuid = _get_uuid(request)
|
|
log.error(f"HTTPException -- uuid: {uuid} | status: {exc.status_code} | msg: {exc.detail}")
|
|
message = _FRIENDLY_MESSAGES.get(exc.status_code, str(exc.detail))
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content=StandardResponse.fail(msg=message, code=exc.status_code, uuid=uuid).model_dump(),
|
|
)
|
|
|
|
|
|
async def validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
|
|
uuid = _get_uuid(request)
|
|
errors = exc.errors()
|
|
log.error(f"ValidationError -- uuid: {uuid} | errors: {errors}")
|
|
return JSONResponse(
|
|
status_code=422,
|
|
content=StandardResponse.fail(msg="数据验证失败", code=422, data=errors, uuid=uuid).model_dump(),
|
|
)
|
|
|
|
|
|
async def assertion_error_handler(request: Request, exc: AssertionError) -> JSONResponse:
|
|
uuid = _get_uuid(request)
|
|
log.error(f"AssertionError -- uuid: {uuid} | msg: {exc}\n{traceback.format_exc()}")
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content=StandardResponse.fail(msg=f"断言错误: {exc}", uuid=uuid).model_dump(),
|
|
)
|
|
|
|
|
|
async def global_exception_handler(request: Request, exc: Exception) -> JSONResponse:
|
|
uuid = _get_uuid(request)
|
|
log.error(f"Unhandled Exception -- uuid: {uuid} | msg: {exc}\n{traceback.format_exc()}")
|
|
msg = str(exc) if settings.env == "dev" else "服务器内部错误"
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content=StandardResponse.fail(msg=msg, uuid=uuid).model_dump(),
|
|
)
|
|
|
|
|
|
def register_exception_handlers(app) -> None:
|
|
"""将异常处理器挂载到 FastAPI 应用"""
|
|
app.add_exception_handler(StarletteHTTPException, http_exception_handler)
|
|
app.add_exception_handler(RequestValidationError, validation_exception_handler)
|
|
app.add_exception_handler(AssertionError, assertion_error_handler)
|
|
app.add_exception_handler(Exception, global_exception_handler)
|