统一响应和异常处理

企业项目里，前端通常希望所有接口返回同一种结构。

常见格式：

{
    "code": 0,
    "data": {},
    "msg": "success"
}

这一节实现：

success() / fail() 统一响应。
ApiResponse[T] 让 Swagger 正确展示响应结构。
BusinessError 表示业务失败。
全局异常处理器统一错误格式。

一、统一响应函数

app/core/response.py：

from typing import Any


def api_response(code: int = 0, data: Any = None, msg: str = "success") -> dict[str, Any]:
    # 全项目统一响应格式：{code,data,msg}。
    return {"code": code, "data": data, "msg": msg}


def success(data: Any = None, msg: str = "success", code: int = 0) -> dict[str, Any]:
    return api_response(code=code, data=data, msg=msg)


def fail(msg: str = "error", code: int = 1, data: Any = None) -> dict[str, Any]:
    return api_response(code=code, data=data, msg=msg)

约定：

字段	含义
`code=0`	业务成功
`code=1`	业务失败
`data`	业务数据
`msg`	给前端展示或调试的消息

二、统一响应模型

app/schemas/response_schema.py：

from typing import Generic, TypeVar

from pydantic import BaseModel, Field

DataT = TypeVar("DataT")


class ApiResponse(BaseModel, Generic[DataT]):
    code: int = Field(title="业务状态码", examples=[0])
    data: DataT | None = Field(default=None, title="响应数据")
    msg: str = Field(title="提示信息", examples=["success"])

路由里这样用：

router.add_api_route(
    "/users",
    list_users_controller,
    methods=["GET"],
    response_model=ApiResponse[UserListResponse],
)

Swagger 会显示：

{
    "code": 0,
    "data": {
        "total": 100,
        "items": []
    },
    "msg": "success"
}

三、业务异常

app/core/business.py：

class BusinessError(Exception):
    # 业务失败：账号已存在、账号密码错误、库存不足等。
    def __init__(self, msg: str):
        self.msg = msg
        super().__init__(msg)

业务层使用：

from app.core.business import BusinessError


def register(...):
    if account_exists:
        raise BusinessError("账号已存在")

业务失败和系统异常要分开：

类型	示例	建议处理
业务失败	账号已存在、密码错误	返回统一业务失败
参数错误	字段缺失、类型错误	返回参数校验失败
权限错误	未登录、Token 过期	HTTP 401
系统异常	数据库断开、代码错误	HTTP 500

四、注册异常处理器

app/core/exceptions.py：

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from sqlalchemy.exc import SQLAlchemyError
from starlette.requests import Request

from app.core.business import BusinessError
from app.core.config import settings
from app.core.response import fail


def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(BusinessError)
    async def business_exception_handler(request: Request, exc: BusinessError):
        return JSONResponse(
            status_code=200,
            content=fail(msg=exc.msg),
        )

    @app.exception_handler(HTTPException)
    async def http_exception_handler(request: Request, exc: HTTPException):
        return JSONResponse(
            status_code=exc.status_code,
            content=fail(msg=str(exc.detail)),
        )

    @app.exception_handler(RequestValidationError)
    async def validation_exception_handler(request: Request, exc: RequestValidationError):
        return JSONResponse(
            status_code=200,
            content=fail(msg="参数校验失败", data=exc.errors()),
        )

    @app.exception_handler(SQLAlchemyError)
    async def sqlalchemy_exception_handler(request: Request, exc: SQLAlchemyError):
        return JSONResponse(
            status_code=500,
            content=fail(msg="数据库异常", data=_debug_data(exc)),
        )

    @app.exception_handler(Exception)
    async def global_exception_handler(request: Request, exc: Exception):
        return JSONResponse(
            status_code=500,
            content=fail(msg="系统异常", data=_debug_data(exc)),
        )


def _debug_data(exc: BaseException) -> dict | None:
    if not settings.debug:
        return None

    return {
        "error_type": exc.__class__.__name__,
        "error": str(exc),
    }

这里的策略和参考项目一致：业务失败、参数校验失败仍返回 HTTP 200，前端通过 code=1 判断。

这是一种常见业务约定，不是 HTTP 唯一正确答案。更偏 REST 的项目也可以让参数校验保持 422，让业务冲突返回 409。团队统一即可。

五、在 app 中注册

app/main.py：

from app.core.exceptions import register_exception_handlers


def create_app() -> FastAPI:
    app = FastAPI(...)

    register_exception_handlers(app)

    return app

异常处理器要在应用创建时注册一次。

六、哪些地方抛什么异常

场景	写法
账号已存在	`raise BusinessError("账号已存在")`
账号密码错误	`raise BusinessError("账号或密码错误")`
未登录	`raise HTTPException(status_code=401, detail="未登录")`
无权限	`raise HTTPException(status_code=403, detail="无权限")`
资源不存在	`raise HTTPException(status_code=404, detail="资源不存在")`
数据库异常	不手动吞掉，交给全局异常处理

不要在每个 controller 里都写一堆 try/except。能统一处理的错误，交给全局异常处理器。

#统一响应和异常处理

#一、统一响应函数

#二、统一响应模型

#三、业务异常

#四、注册异常处理器

#五、在 app 中注册

#六、哪些地方抛什么异常