70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
"""统一响应 + 全局异常处理。所有接口返回 {code, message, data}。"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI, Request, status
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.exceptions import HTTPException as StarletteHTTPException
|
|
|
|
from app.core.logging import logger
|
|
|
|
|
|
class AppError(Exception):
|
|
"""业务异常基类。服务层抛出，由全局 handler 转成标准响应体。"""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
*,
|
|
code: int = 1000,
|
|
http_status: int = status.HTTP_400_BAD_REQUEST,
|
|
data: Any = None,
|
|
) -> None:
|
|
super().__init__(message)
|
|
self.message = message
|
|
self.code = code
|
|
self.http_status = http_status
|
|
self.data = data
|
|
|
|
|
|
def ok(data: Any = None, message: str = "ok") -> dict:
|
|
return {"code": 0, "message": message, "data": data}
|
|
|
|
|
|
def fail(code: int, message: str, data: Any = None) -> dict:
|
|
return {"code": code, "message": message, "data": data}
|
|
|
|
|
|
def register_exception_handlers(app: FastAPI) -> None:
|
|
@app.exception_handler(AppError)
|
|
async def _app_error(request: Request, exc: AppError) -> JSONResponse:
|
|
logger.warning("app_error", path=request.url.path, code=exc.code, message=exc.message)
|
|
return JSONResponse(
|
|
status_code=exc.http_status,
|
|
content=fail(exc.code, exc.message, exc.data),
|
|
)
|
|
|
|
@app.exception_handler(StarletteHTTPException)
|
|
async def _http_exc(request: Request, exc: StarletteHTTPException) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content=fail(exc.status_code, str(exc.detail)),
|
|
)
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def _validation_exc(request: Request, exc: RequestValidationError) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
content=fail(422, "参数校验失败", exc.errors()),
|
|
)
|
|
|
|
@app.exception_handler(Exception)
|
|
async def _unhandled(request: Request, exc: Exception) -> JSONResponse:
|
|
logger.exception("unhandled_exception", path=request.url.path)
|
|
return JSONResponse(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
content=fail(500, "服务器内部错误"),
|
|
)
|