第 8 章:错误处理与异常体系
业务代码里
return {"code": 500, "msg": "..."}是最常见的坏习惯。本章把它拆掉,换成一整套基于 HTTP 语义、客户端可统一处理的异常体系。
学习目标
- 理解 HTTP 状态码语义丢失为什么会让客户端无法统一处理错误
- 掌握
HTTPException的三个参数与fastapi.status常量的用法 - 能定义自定义业务异常并注册处理器,输出项目统一的错误响应体
- 掌握改造
RequestValidationError与StarletteHTTPException的方法 - 理解异常处理器的注册位置与中间件中的异常传播链路
8.1 反模式先行:为什么不能 return {"code": 500}
❌ 不推荐:
@app.get("/users/{user_id}")
async def get_user(user_id: int) -> dict[str, object]:
if user_id == 999:
return {"code": 404, "msg": "用户不存在"} # HTTP 状态码仍是 200
return {"code": 200, "data": {"id": user_id, "name": "moqian"}}这段代码能跑,但代价很大:
| 问题 | 后果 |
|---|---|
HTTP 状态码永远是 200 | 网关、CDN、监控、重试策略全部失效;5xx 告警不会触发 |
| 客户端必须解析 body 才知道成败 | 无法用统一的 if (!response.ok) 处理,每个接口都要写一遍判断 |
| OpenAPI 文档失真 | 文档里只声明了 200,客户端代码生成器不会生成错误分支 |
| 中间件无法介入 | 鉴权、限流、重试类中间件按状态码决策,全部失灵 |
| 与框架自身的错误格式冲突 | 校验失败返回 {"detail": [...]},业务失败返回 {"code": ...},两套格式 |
✅ 推荐:传输层的错误用 HTTP 状态码表达,业务错误用业务错误码表达,两者并存。
from fastapi import HTTPException, status
@app.get("/users/{user_id}")
async def get_user(user_id: int) -> dict[str, object]:
if user_id == 999:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在")
return {"id": user_id, "name": "moqian"}8.2 HTTPException 详解
from fastapi import HTTPException, status
async def protected() -> dict[str, str]:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="凭证无效或已过期",
headers={"WWW-Authenticate": "Bearer"},
)| 参数 | 说明 |
|---|---|
status_code | HTTP 状态码,建议用 fastapi.status 常量(如 status.HTTP_404_NOT_FOUND),不用魔术数字 |
detail | 错误详情,默认响应体为 {"detail": ...},可以是任意可 JSON 序列化的值(如列表) |
headers | 附加响应头,WWW-Authenticate 是 401 的标准用法 |
8.3 自定义业务异常
HTTPException 适合 HTTP 语义清晰的错误;业务层往往需要更细的分类(余额不足、手机号已注册)。做法是定义异常类 + 注册处理器。
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
class BusinessError(Exception):
"""所有业务异常的基类。"""
code: str = "BUSINESS_ERROR"
status_code: int = status.HTTP_400_BAD_REQUEST
def __init__(self, message: str) -> None:
self.message = message
super().__init__(message)
class UserNotFoundError(BusinessError):
code = "USER_NOT_FOUND"
status_code = status.HTTP_404_NOT_FOUND
app = FastAPI()
@app.exception_handler(BusinessError)
async def business_error_handler(request: Request, exc: BusinessError) -> JSONResponse:
return JSONResponse(
status_code=exc.status_code,
content={"code": exc.code, "message": exc.message, "detail": None},
)
@app.get("/users/{user_id}")
async def get_user(user_id: int) -> dict[str, object]:
if user_id == 999:
raise UserNotFoundError("用户 999 不存在")
return {"id": user_id, "name": "moqian"}⚠️ 继承关系很重要:
@app.exception_handler(BusinessError)会捕获所有子类,父类处理器兜底;若同时注册了具体子类的处理器,Starlette 按MRO顺序优先匹配具体类型。需要携带结构化细节时,在子类的__init__里挂上self.detail = detail or [],处理器里一并输出即可。
8.4 统一错误响应体设计
{
"code": "USER_NOT_FOUND",
"message": "用户不存在",
"detail": [{ "field": "user_id", "message": "该 ID 无对应记录", "type": "not_found" }]
}| 字段 | 含义 | 客户端用法 |
|---|---|---|
code | 机器可读的稳定错误码,永不变更 | switch (code) 做本地化与特定处理 |
message | 面向开发者的简短描述 | 日志、调试 |
detail | 可选的结构化明细(校验失败时逐字段列出) | 表单逐项高亮 |
🔑 关键原则:
code是契约,message是给人看的。永远不要让客户端去匹配message文案,一旦改文案就会破坏兼容性。
8.5 覆盖 RequestValidationError
FastAPI 对 Pydantic 校验失败默认返回 422,结构是 {"detail": [ ... ]}。改造成项目统一格式:
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
request: Request, exc: RequestValidationError
) -> JSONResponse:
details = [
{
"field": ".".join(str(part) for part in error["loc"]),
"message": error["msg"],
"type": error["type"],
# ctx 里可能含 ValueError 等不可序列化对象,统一降级
"ctx": jsonable_encoder(error.get("ctx"), custom_encoder={Exception: str}),
}
for error in exc.errors()
]
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content={"code": "VALIDATION_ERROR", "message": "请求参数校验失败", "detail": details},
)errors() 中每个条目的关键键:
| 键 | 示例 | 说明 |
|---|---|---|
loc | ("body", "password", 0) | 错误位置的元组,从来源开始逐层定位 |
msg | "String should have at least 8 characters" | 人类可读的错误描述 |
type | "string_too_short" | 机器可读的错误类型,适合客户端映射 |
ctx | {"min_length": 8} | 错误上下文,可能包含非 JSON 序列化对象 |
input | 用户传入的原始值 | 可能很大或含敏感信息,不建议回显 |
⚠️ 最大的坑:
ctx里可能塞着ValueError实例或其他不可序列化对象(自定义校验器抛错时尤其常见)。直接JSONResponse(content={"detail": exc.errors()})会在运行时报TypeError: Object of type ValueError is not JSON serializable,所以必须先过一层编码处理。
8.5.1 RequestValidationError 与 ValidationError 别混淆
| 类型 | 来源 | 导入位置 |
|---|---|---|
RequestValidationError | FastAPI 解析请求(路径/查询/头/Cookie/body)失败 | fastapi.exceptions |
pydantic.ValidationError | 业务代码里手动调用 Model.model_validate() 失败 | pydantic |
导入错了类,处理器永远不会被触发。业务层建议把 Pydantic 的异常转换成自定义 BusinessError,只维护一套响应格式:
def parse_or_raise(raw: dict[str, object]) -> Payload:
try:
return Payload.model_validate(raw)
except ValidationError as exc: # 注意:这里是 pydantic.ValidationError
raise BusinessError(f"内部数据不合法:{exc.error_count()} 处错误") from exc8.6 覆盖 StarletteHTTPException
404 Not Found、405 Method Not Allowed 这类由 Starlette 路由层直接抛出的异常(是 FastAPI HTTPException 的父类)不会走子类处理器。要统一格式必须处理基类:
from starlette.exceptions import HTTPException as StarletteHTTPException
_STATUS_CODE_MAP = {404: "NOT_FOUND", 405: "METHOD_NOT_ALLOWED", 500: "INTERNAL_SERVER_ERROR"}
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(
request: Request, exc: StarletteHTTPException
) -> JSONResponse:
return JSONResponse(
status_code=exc.status_code,
content={
"code": _STATUS_CODE_MAP.get(exc.status_code, "HTTP_ERROR"),
"message": str(exc.detail),
"detail": None,
},
headers=getattr(exc, "headers", None),
)🔑 必须回传
exc.headers,否则WWW-Authenticate、Allow这类标准响应头会丢,浏览器与客户端的标准行为会被破坏。
至此三条链路全部统一:业务异常、请求校验失败、框架层 HTTP 异常,客户端只需处理一种结构。
8.7 处理器注册位置与工程组织
| 注册方式 | 写法 | 适用 |
|---|---|---|
| 装饰器 | @app.exception_handler(Exc) | 可读性好,适合就近定义 |
| 集中注册 | app.add_exception_handler(Exc, func) | 在工厂函数/装配模块里按条件注册 |
APIRouter 上不能注册全局处理器(没有 exception_handler,会报 AttributeError)。正确做法是把三个处理器集中到一个模块,只暴露一个装配函数:
# app/core/exceptions.py
from fastapi import FastAPI
def register_exception_handlers(app: FastAPI) -> None:
"""在 create_app() 里调用一次;三个处理器的实现分别见 8.3 / 8.5 / 8.6 节。"""
app.add_exception_handler(BusinessError, business_error_handler)
app.add_exception_handler(RequestValidationError, validation_exception_handler)
app.add_exception_handler(StarletteHTTPException, http_exception_handler)在 app/main.py 里 app = FastAPI(title="Blog API") 之后调用 register_exception_handlers(app)。这样第 11 章的分层结构里,任何 service 层都可以 raise BusinessError(...),不需要 import FastAPI,错误格式也只有一个来源。
8.8 异常传播链路与中间件
要点:
- 处理器优先于中间件:已注册处理器的异常由
ExceptionMiddleware在用户中间件内层处理,用户中间件看到的是正常响应。 - 未捕获异常冒泡到
ServerErrorMiddleware(Starlette 最外层),返回500,此时你注册的处理器不生效。 - 500 响应体是固定的(
"Internal Server Error"),不含堆栈——这是安全默认值,不要关掉。
若确实需要在中间件里兜底,用 try/except 包住 call_next,在 except 里 logger.exception(...) 记录堆栈,然后返回 {"code": "INTERNAL_ERROR", "message": "服务器内部错误"}——堆栈只进日志,不进响应。注意这会绕开 ServerErrorMiddleware 的语义,非必要不用。
8.9 生产环境的错误信息粒度
| 信息 | 开发环境 | 生产环境 |
|---|---|---|
| 堆栈跟踪 / SQL 语句 | 控制台可见 | 绝不返回客户端,只进日志(SQL 还含表结构与数据) |
| 文件路径 / 主机名 / 依赖版本 | 可打印 | 不返回,泄露部署结构与可被利用的 CVE 线索 |
| 业务错误码与提示 | 返回 | 返回,这是契约 |
| 请求 ID | 返回 | 返回,便于对账定位 |
❌ 不推荐:
except Exception as exc:
return JSONResponse(status_code=500, content={"error": str(exc), "traceback": _format_exc()})✅ 推荐:
import uuid
async def safe_error_response(request: Request, exc: Exception) -> dict[str, str]:
request_id = getattr(request.state, "request_id", str(uuid.uuid4()))
logger.exception("请求处理失败 request_id=%s path=%s", request_id, request.url.path)
# 客户端只拿到稳定的错误码与可对账的 request_id
return {"code": "INTERNAL_ERROR", "message": "服务器内部错误", "request_id": request_id}第 9 章会把 request_id 注入做成中间件,让「响应头里的 ID」与「用户看到的错误」对得上。
常见坑与排查
| 现象 | 原因 | 解决 |
|---|---|---|
裸 HTTPException 的 {"detail": ...} 与项目 {"code": ...} 格式不一致 | 只覆盖了部分链路,或自定义处理器漏掉了 HTTPException | 同时注册 RequestValidationError + StarletteHTTPException,并让自定义异常汇聚到同一个构造函数 |
404 / 405 仍返回 {"detail": "Not Found"} | FastAPI 的 HTTPException 是 Starlette 的子类,只注册子类接不住路由层抛出的基类异常 | 注册 starlette.exceptions.HTTPException(基类) |
RequestValidationError 处理器不生效 | 导入成了 pydantic.ValidationError | 用 from fastapi.exceptions import RequestValidationError |
| 处理器里再抛异常,客户端收到 500 | 处理器自身有 bug(如访问了不存在的字段),异常逃逸到 ServerErrorMiddleware | 处理器内只用 exc 上确定存在的属性,并对 detail 做空值兜底 |
TypeError: Object of type ValueError is not JSON serializable | exc.errors() 的 ctx 里有自定义校验器抛出的异常对象 | 序列化前用 jsonable_encoder(..., custom_encoder={Exception: str}) 降级 |
401 响应缺 WWW-Authenticate,浏览器不弹认证框 | 自定义处理器构造 JSONResponse 时丢了 exc.headers | headers=getattr(exc, "headers", None) |
在 APIRouter 上注册处理器报 AttributeError | APIRouter 不支持全局异常处理器 | 集中到 register_exception_handlers(app),在 create_app() 里调用 |
本章小结
| 要点 | 说明 |
|---|---|
| 反模式 | return {"code": 500} 会丢失 HTTP 语义,破坏网关、监控与文档 |
HTTPException | status_code + detail + headers,默认响应 {"detail": ...};状态码用 fastapi.status 常量 |
| 业务异常 | 定义基类 + 子类携带 code / status_code,一个处理器接住全部 |
| 统一响应体 | code(稳定机器码)+ message(给人看)+ detail(结构化明细) |
| 校验错误 | 覆盖 RequestValidationError,注意 ctx 可能不可 JSON 序列化 |
两个 ValidationError | 请求校验用 RequestValidationError,手动校验用 pydantic.ValidationError |
| 框架异常 | 覆盖 starlette.exceptions.HTTPException 才能接住 404 / 405,且要回传 headers |
| 工程组织 | app/core/exceptions.py 暴露 register_exception_handlers(app) |
| 异常层级 | ExceptionMiddleware 在内层处理已注册异常,未捕获的冒泡到 ServerErrorMiddleware 变成 500 |
| 生产粒度 | 不回显堆栈、SQL、路径;返回稳定的 code 与 request_id |
练习题
- 定义
InsufficientBalanceError(余额不足,携带required与available两个数值),继承BusinessError,并输出带detail的统一响应体。 - 把 8.5 节的校验处理器改造成「只返回前 5 条错误」,并在超过 5 条时追加一条
{"field": "_", "message": "还有 N 条错误未展示"}。 - 写一段代码复现
TypeError: Object of type ValueError is not JSON serializable:自定义一个field_validator抛ValueError,观察ctx的真实内容,再修复它。
下一章预告
错误体系解决的是「一段请求内怎么报错」。接下来把视角拉到请求之外:请求进来先经过谁、应用启动时要建什么、响应返回后还能做什么。