Become a sponsor

全局异常与响应
定义了统一的响应格式和全局异常处理机制。所有 API 返回 {code, data, msg, ok} 标准格式,异常被捕获后也转换为相同格式。响应函数位于 src/core/response.py,异常类型位于 src/core/exceptions.py。
src/core/response.py 提供统一的响应生成函数:
from core import response as R
# 成功响应
R.ok(data={"id": 1}, msg="操作成功")
# {"code": 0, "data": {"id": 1}, "msg": "操作成功", "ok": true}
# 成功响应(带额外字段)
R.ok(data=list, count=total)
# {"code": 0, "data": [...], "msg": "操作成功", "ok": true, "count": 100}
# 失败响应
R.failed(msg="参数错误")
# {"code": 1, "data": null, "msg": "参数错误", "ok": false}| 函数 | 说明 | code | ok |
|---|---|---|---|
R.ok() | 成功响应 | 0 | true |
R.failed() | 失败响应 | 1 | false |
R.page() | 分页响应 | 0 | true |
R.response() | 通用响应 | 自定义 | 自定义 |
函数签名:
def ok(data=None, msg="操作成功", code=0, **kwargs) -> JSONResponse:
"""生成成功响应,支持通过 kwargs 添加额外字段(如 count)"""
def failed(msg="操作失败", code=1, data=None, **kwargs) -> JSONResponse:
"""生成失败响应"""
def page(data: list, total: int, current: int, size: int, msg="操作成功", code=0) -> JSONResponse:
"""分页响应:封装 {records, total, size, current, pages} 标准分页结构"""
pages = (total + size - 1) // size if size > 0 else 0
def response(data=None, msg="操作成功", code=0, success=True, **kwargs) -> JSONResponse:
"""通用响应生成器(高级用法),允许完全自定义字段"""R.page(data=records, total=100, current=1, size=10)
# {
# "code": 0,
# "data": {
# "records": [...],
# "total": 100,
# "size": 10,
# "current": 1,
# "pages": 10
# },
# "msg": "操作成功",
# "ok": true
# }认证异常,由 login_required 中间件抛出:
# src/core/exceptions.py
class AuthorizationException(Exception):
def __init__(self, code: int, msg: str):
self.code = code
self.msg = msg业务异常,业务代码可主动抛出:
class BusinessException(Exception):
def __init__(self, code: int = 1, msg: str = ""):
self.code = code
self.msg = msg使用示例:
from core.exceptions import BusinessException
def some_business_logic():
if some_condition:
raise BusinessException(code=1, msg="数据不存在")src/core/app.py 注册了全局异常处理器,捕获所有异常并转换为标准格式:
def register_exception(app: FastAPI):
# 认证异常
@app.exception_handler(AuthorizationException)
async def auth_exception_handler(request, exc):
return JSONResponse(
status_code=200,
content={"code": exc.code, "data": None, "msg": exc.msg, "ok": False},
)
# 业务异常
@app.exception_handler(BusinessException)
async def business_exception_handler(request, exc):
return JSONResponse(
status_code=200,
content={"code": exc.code, "data": None, "msg": exc.msg, "ok": False},
)
# HTTP 异常(404/405 等)
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
return JSONResponse(
status_code=200,
content={"code": 1, "data": None, "msg": exc.detail, "ok": False},
)
# 参数验证异常
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
errors = []
for err in exc.errors():
field = err.get("loc", [""])[-1]
msg = err.get("msg", "")
errors.append(f"{field}: {msg}" if field and field != "body" else msg)
return JSONResponse(
status_code=200,
content={"code": 1, "data": None, "msg": "/".join(errors)},
)
# 未捕获异常
@app.exception_handler(Exception)
async def exception_handler(request, exc):
return JSONResponse(
status_code=200,
content={"code": 1, "data": None, "msg": "服务器内部错误"},
)HTTP 200
所有异常响应均返回 HTTP 200 状态码,业务状态通过 code 字段区分。这是前后端分离架构的常见做法,避免浏览器拦截非200响应。
# src/core/response.py
class Codes:
SUCCESS = 0
FAILED = 1
UNAUTHORIZED = 401 # 未认证
FORBIDDEN = 403 # 无权限
NOT_FOUND = 404 # 资源不存在
VALIDATE_ERROR = 422 # 验证错误
SERVER_ERROR = 500 # 服务器错误全局异常与响应模块具备以下特点:
1. 统一格式:所有 API 返回 {code, data, msg, ok} 标准格式
2. 全局捕获:认证异常、业务异常、HTTP异常、验证异常、未捕获异常
3. HTTP 200:所有响应均返回 200,业务状态通过 code 区分
4. 错误聚合:多个验证错误用 / 分隔,便于前端展示
5. 安全隐藏:未捕获异常返回通用提示,不暴露内部错误信息