1. 失败
一次接口调用失败,原因不一定都在同一层。把所有错误都返回成同一个状态码或同一句文本,会让客户端无法判断应该修改请求、重新登录、稍后重试,还是联系服务维护者。
可以先把失败分为三类:
| 类型 | 示例 | 主要处理方 |
|---|---|---|
| 请求错误 | 缺少字段、类型错误、认证信息无效 | 客户端修改请求或身份信息 |
| 业务错误 | 会话不存在、余额不足、状态冲突 | 客户端按业务状态处理 |
| 系统错误 | 代码异常、数据库断开、上游不可用 | 服务端记录、告警和修复 |
这三类失败最终都可以表现为 HTTP 响应,但处理策略不同。HTTP 状态码描述协议层结果,业务错误代码供程序稳定判断,日志和 Request ID 则帮助服务端定位问题。它们各自解决不同问题,不能互相替代。
2. HTTPException
对于已经预料到、并且能够明确转换成 HTTP 响应的失败,可以抛出 HTTPException:
01from fastapi import FastAPI, HTTPException, status0203app = FastAPI()0405conversations = {"conversation-1": {"title": "FastAPI"}}0607@app.get("/conversations/{conversation_id}")08async def get_conversation(conversation_id: str):09conversation = conversations.get(conversation_id)1011if conversation is None:12raise HTTPException(13status_code=status.HTTP_404_NOT_FOUND,14detail="会话不存在",15)1617return conversation
这里必须使用 raise,而不是 return。异常抛出后,当前调用链停止执行,FastAPI 再把它转换成 HTTP 响应。因此,路由函数调用的依赖或普通工具函数也可以抛出 HTTPException,不需要层层返回错误标记。
默认响应如下:
1{2"detail": "会话不存在"3}
detail 不只支持字符串,也支持字典、列表等能够编码成 JSON 的数据。HTTPException 还可以通过 headers 返回错误响应头。例如,401 响应通常需要携带 WWW-Authenticate,429 或 503 响应可以在适用时携带 Retry-After。
HTTPException 适合表达已知的 HTTP 失败,不适合掩盖程序缺陷。数据库驱动抛出的异常、空指针和未知逻辑错误不应该被随手包装成 400;这些错误需要保留堆栈并进入统一的 500 处理流程。
3. 状态码
选择状态码时,要表达失败发生在哪个边界,而不是只看错误文案:
| 状态码 | 常见使用场景 |
|---|---|
| 400 | 通用的客户端请求错误,并且没有更准确的状态码可用 |
| 401 | 缺少或提供了无效认证凭证,通常同时返回 WWW-Authenticate |
| 403 | 身份已经确认,但不允许执行当前操作 |
| 404 | 路由或目标资源不存在;有时也用于隐藏无权访问资源是否存在 |
| 405 | 路径存在,但当前 HTTP 方法不被允许 |
| 409 | 当前资源状态发生冲突,例如重复创建或版本冲突 |
| 422 | 请求内容可以解析,但字段类型、约束或组合规则没有通过校验 |
| 429 | 调用方超过本服务的速率或配额限制 |
| 500 | 未预期的服务端错误 |
| 502 | 作为网关访问上游时,收到了无效响应 |
| 503 | 服务暂时不可用、过载或正在维护 |
| 504 | 作为网关访问上游时,等待响应超时 |
400 是较宽泛的客户端错误,422 则更明确地表示:服务已经理解请求的媒体类型与语法,但内容没有通过校验。FastAPI 对路径参数、查询参数和请求体的默认校验失败响应就是 422。
401 与 403 也经常被混淆。没有有效身份时使用 401;身份有效但权限不足时使用 403。不要为了让前端始终进入“成功分支”而全部返回 200,HTTP 客户端、缓存、代理和监控都依赖真实状态码。
4. 业务异常
小型接口可以直接抛出 HTTPException,但随着业务增多,如果领域服务到处依赖 FastAPI,业务规则就会和传输协议绑定在一起。更稳妥的做法是先定义领域异常,再在 HTTP 边界统一转换。
1class ConversationNotFoundError(Exception):2def __init__(self, conversation_id: str):3self.conversation_id = conversation_id4super().__init__(f"conversation not found: {conversation_id}")
业务服务只表达发生了什么:
1from domain_errors import ConversationNotFoundError23async def find_conversation(conversation_id: str):4conversation = await repository.find_by_id(conversation_id)56if conversation is None:7raise ConversationNotFoundError(conversation_id)89return conversation
FastAPI 应用负责把它翻译成 HTTP 响应:
01from fastapi import FastAPI, Request, status02from fastapi.responses import JSONResponse0304from domain_errors import ConversationNotFoundError0506app = FastAPI()0708@app.exception_handler(ConversationNotFoundError)09async def handle_conversation_not_found(10request: Request,11error: ConversationNotFoundError,12) -> JSONResponse:13request_id = getattr(request.state, "request_id", None)1415return JSONResponse(16status_code=status.HTTP_404_NOT_FOUND,17content={18"error": {19"code": "conversation_not_found",20"message": "会话不存在",21"request_id": request_id,22}23},24)
这样,业务服务只关心“没有找到会话”,HTTP 层负责决定它对应 404 和怎样的 JSON 结构。如果同一套业务逻辑以后被命令行程序、定时任务或消息消费者调用,领域异常仍然可以复用。
实际项目中可以按业务模块定义少量有明确语义的异常,例如 ConversationNotFoundError、PermissionDeniedError 和 StateConflictError。不要为每一句错误文案都创建一个异常类,否则异常体系会比业务本身更难维护。
5. 错误协议
接口只返回自然语言时,前端只能匹配文案。更稳定的错误协议应该同时包含机器可读代码和面向人的说明:
1{2"error": {3"code": "conversation_not_found",4"message": "会话不存在",5"request_id": "req_abc123",6"details": []7}8}
字段职责如下:
code:稳定的业务标识,前端据此决定交互分支;message:面向用户或开发者的说明,可以调整和国际化;request_id:关联本次请求的日志、指标和调用链;details:可选的结构化细节,常用于字段校验错误。
并不是每个响应都必须带空的 details。关键是同一项目要约定统一的外层结构,以及每个错误代码对应的 HTTP 状态码。前端不要通过匹配中文消息判断错误类型,日志系统也不要拿 message 代替稳定的错误代码做聚合。
Request ID 通常由中间件在请求进入时生成或接收,再写入 request.state 和响应头。异常处理器只负责读取它,不能在不同处理器中各自生成,否则同一次请求的日志会出现多个 ID。
6. 校验错误
路径参数、查询参数、请求头和请求体无法通过校验时,FastAPI 会抛出 RequestValidationError,默认返回 422。错误列表通常包含以下信息:
loc:错误位于path、query、header或body的哪个字段;type:Pydantic 生成的错误类型;msg:便于阅读的错误说明;input:导致失败的原始输入;ctx:某些规则附带的校验上下文。
如果项目需要统一外层协议,可以注册自己的处理器。下面只公开字段位置、类型和说明,没有把原始输入直接返回给客户端:
01from fastapi import FastAPI, Request, status02from fastapi.exceptions import RequestValidationError03from fastapi.responses import JSONResponse0405app = FastAPI()0607@app.exception_handler(RequestValidationError)08async def handle_validation_error(09request: Request,10error: RequestValidationError,11) -> JSONResponse:12request_id = getattr(request.state, "request_id", None)13fields = [14{15"location": list(item["loc"]),16"type": item["type"],17"message": item["msg"],18}19for item in error.errors()20]2122return JSONResponse(23status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,24content={25"error": {26"code": "validation_error",27"message": "请求数据校验失败",28"request_id": request_id,29"details": fields,30}31},32)
RequestValidationError 还保存了 body。它在本地调试时很有用,但正文、密码、Token 或上传内容可能包含敏感数据,生产环境不要默认把 error.body 写进响应或完整日志。errors() 中的 input 和 ctx 也应该经过审查后再公开,有些上下文对象还不一定能够直接 JSON 序列化。
这里还要区分两种异常:
RequestValidationError表示客户端请求没有通过 FastAPI 的输入校验,通常对应 422;- 服务内部出现的 Pydantic
ValidationError可能意味着代码构造了不符合模型的数据,通常应该作为服务端错误调查,而不是伪装成客户端的 422。
如果只想在校验失败时补充日志,并不打算改变 FastAPI 的默认响应,可以从 fastapi.exception_handlers 导入并调用 request_validation_exception_handler,没有必要复制一份默认实现。
7. 处理层级
理解异常经过哪些层,能够避免“明明注册了处理器却没有得到预期响应”的问题。Starlette 应用的核心处理顺序可以简化为:
| 层级 | 主要职责 |
|---|---|
ServerErrorMiddleware | 捕获未处理的服务端错误并生成 500 响应 |
| 自定义中间件 | 日志、Request ID、CORS、耗时统计等横切逻辑 |
ExceptionMiddleware | 把 HTTPException 等已处理异常转换成响应 |
| Router 与 Endpoint | 匹配路由,执行依赖和路径函数 |
HTTPException 属于“可预期、可转换”的异常,由 ExceptionMiddleware 处理后成为普通响应;未知异常则会继续向外传播,最终进入服务端错误处理流程。因此,日志中间件遇到未知异常时应该记录后重新抛出,不能静默吞掉。
针对具体异常类型注册的处理器会沿异常类的继承关系匹配。RequestValidationError 可以使用自己的处理器,业务异常可以使用各自的处理器,而 Exception 只负责最后兜底。404、405 等框架生成的 HTTP 错误仍然属于 Starlette 的 HTTPException。
还有一个容易忽略的边界:HTTPException 适合在路由、依赖以及它们调用的业务函数中抛出;在 ASGI 中间件中需要提前拒绝请求时,应直接返回相应的 Response。
8. HTTP 处理器
如果要统一所有 HTTP 错误的外层结构,处理器应该注册到 Starlette 的 HTTPException。这是因为 FastAPI 的版本继承自 Starlette 的版本,而框架内部的 404、405 等错误可能直接使用 Starlette 的类型。
01from fastapi import FastAPI, Request02from fastapi.encoders import jsonable_encoder03from fastapi.responses import JSONResponse04from starlette.exceptions import HTTPException as StarletteHTTPException0506app = FastAPI()0708@app.exception_handler(StarletteHTTPException)09async def handle_http_error(10request: Request,11error: StarletteHTTPException,12) -> JSONResponse:13request_id = getattr(request.state, "request_id", None)1415if isinstance(error.detail, str):16content = {17"error": {18"code": f"http_{error.status_code}",19"message": error.detail,20"request_id": request_id,21}22}23else:24content = {25"error": {26"code": f"http_{error.status_code}",27"message": "请求失败",28"request_id": request_id,29"details": jsonable_encoder(error.detail),30}31}3233return JSONResponse(34status_code=error.status_code,35headers=error.headers,36content=content,37)
业务代码仍然可以继续抛出 fastapi.HTTPException。统一格式时有两个细节不能遗漏:
- 保留
error.headers,否则WWW-Authenticate、Retry-After等协议字段会丢失; - 保留结构化
detail的数据结构,不能使用str(error.detail)把字典变成不可解析的文本。
如果只是想记录 HTTP 错误而不改变默认响应,也可以调用 FastAPI 内置的 http_exception_handler。复用官方处理器比手动复制默认 JSON 格式更稳妥。
9. 未知异常
未预期异常通常应该返回 500,但不能把堆栈、密钥、数据库语句或供应商原始响应发给客户端。可以注册一个最终兜底处理器,对外返回稳定协议,对内保留完整堆栈:
01import logging0203from fastapi import FastAPI, Request, status04from fastapi.responses import JSONResponse0506logger = logging.getLogger(__name__)07app = FastAPI()0809@app.exception_handler(Exception)10async def handle_unexpected_error(11request: Request,12error: Exception,13) -> JSONResponse:14request_id = getattr(request.state, "request_id", None)1516logger.exception(17"Unhandled request error",18extra={19"request_id": request_id,20"method": request.method,21"path": request.url.path,22},23)2425return JSONResponse(26status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,27content={28"error": {29"code": "internal_server_error",30"message": "服务暂时无法完成请求",31"request_id": request_id,32}33},34)
这里的 logger.exception() 会记录异常堆栈,公开响应只提供通用信息。生产日志通常还需要记录路由、请求方法、上游名称和耗时,但不要默认记录 Authorization、Cookie、完整请求体、提示词或文件内容。
不要在每个路由里使用下面这种宽泛捕获:
1try:2return await service.run()3except Exception:4return {"error": "failed"}
它不仅把 HTTP 状态码错误地变成了 200,还丢失了异常堆栈、统一协议和监控信号。只有当当前层真正知道如何恢复、补偿或翻译异常时,才应该捕获它。
开发环境中的 debug=True 会让 500 响应展示调试信息,并绕过自定义 500 页面。它只适合本地调试,生产环境必须关闭。
10. 上游错误
模型服务、数据库代理和外部 HTTP API 会产生超时、限流、认证失败、连接失败和响应协议错误。应用应该在上游适配层识别这些异常,再转换成本系统的稳定语义。
01import asyncio0203from fastapi import HTTPException, status0405async def invoke_with_timeout(model, messages):06try:07async with asyncio.timeout(30):08return await model.ainvoke(messages)09except TimeoutError as error:10raise HTTPException(11status_code=status.HTTP_504_GATEWAY_TIMEOUT,12detail={13"code": "model_timeout",14"message": "模型响应超时",15},16) from error
asyncio.timeout() 从 Python 3.11 开始提供。超出期限时,它会取消上下文中的等待,并在上下文外转换成内置的 TimeoutError,所以上面的 except 必须写在 async with 外侧。
映射上游错误时要保留真实语义:
- 调用方超过本服务为它设置的限额,可以返回 429;
- 共享上游限流导致本服务暂时无法处理请求,可以返回 503;
- 上游连接成功但返回了无效协议内容,可以返回 502;
- 等待上游超过整条请求的期限,可以返回 504;
- 上游认证失败通常是服务端配置问题,不能直接把供应商的密钥错误和响应正文暴露给调用方。
是否重试取决于请求是否幂等、错误是否可恢复以及剩余时间预算。重试还应设置次数上限、指数退避和随机抖动,并在适用时遵循 Retry-After。对所有异常立即重试,反而可能在上游故障时放大流量。
11. 取消
超时与取消相关,但不是同一件事。任务可能因为服务关闭、上层主动取消或并发任务失败而收到 asyncio.CancelledError。取消表示“这项工作已经没有继续执行的必要”,通常应该尽快结束,并把取消信号继续向上传播。
1import asyncio23async def run_model(model, messages):4try:5return await model.ainvoke(messages)6except asyncio.CancelledError:7await release_temporary_resources()8raise
在当前 Python 中,CancelledError 继承自 BaseException,普通的 except Exception 不会捕获它。如果为了清理资源而显式捕获,就必须在清理后再次 raise。把取消转换成 500、504 或一个空结果,会让服务继续占用已经不再需要的资源,也会破坏 TaskGroup、超时和服务关闭流程。
12. 响应时机
异常只有在响应尚未发送时,才能转换成新的 HTTP 状态码和 JSON 正文。流式响应发送第一块数据后再出错,响应头和状态码通常已经确定,此时不能把 200 改成 500。应用只能按约定发送流内错误事件、结束连接并记录日志。
后台任务也发生在响应之后。即使异常处理器收到了后台任务抛出的异常,它生成的新响应也已经无法替换先前发出的响应。后台任务必须自行记录失败、执行有限重试,或把状态写入可查询的存储。多个后台任务按顺序运行时,前一个任务抛出异常还会阻止后续任务执行。
WebSocket 升级完成后也不再使用普通 HTTP 状态码表达错误。连接建立前可以拒绝升级;连接建立后应使用 WebSocketException 或合适的关闭码结束连接。
因此,“异常能否改写 HTTP 响应”取决于错误发生的时间和协议阶段,而不只取决于异常类型。
13. 文档与测试
运行时注册了异常处理器,不代表 OpenAPI 会自动知道每个路由可能返回哪些业务错误。需要通过路径装饰器的 responses 参数声明错误模型:
01from uuid import UUID0203from fastapi import APIRouter04from pydantic import BaseModel0506router = APIRouter()0708class ErrorBody(BaseModel):09code: str10message: str11request_id: str | None = None1213class ErrorResponse(BaseModel):14error: ErrorBody1516@router.get(17"/conversations/{conversation_id}",18responses={19404: {20"model": ErrorResponse,21"description": "会话不存在",22}23},24)25async def get_conversation(conversation_id: UUID):26return await service.find_conversation(str(conversation_id))
responses 负责生成文档,不会替你执行异常转换,也不会在运行时强制校验手动创建的 JSONResponse。实际的异常处理器、错误协议和 OpenAPI 声明必须保持一致。多个路由共用同一组错误时,可以抽取公共 responses 字典再合并使用。
错误分支也需要测试,而且应该断言稳定字段:
01from fastapi.testclient import TestClient0203from main import app0405client = TestClient(app)0607def test_conversation_not_found():08response = client.get(09"/conversations/123e4567-e89b-12d3-a456-426614174000"10)1112assert response.status_code == 40413assert response.json()["error"]["code"] == "conversation_not_found"1415def test_invalid_conversation_id():16response = client.get("/conversations/not-a-uuid")1718assert response.status_code == 42219assert response.json()["error"]["code"] == "validation_error"
测试不必断言整段自然语言,因为文案可能调整。更值得验证的是状态码、错误代码、必要响应头和字段位置。测试未知 500 处理器时,可以创建 TestClient(app, raise_server_exceptions=False),否则测试客户端默认会把服务端异常重新抛给测试进程。
14. 安全
错误信息也是接口的数据边界:
- 不返回 API Key、Authorization Header 和 Cookie;
- 不返回完整提示词、用户隐私和模型原始调试对象;
- 不把数据库语句、内部路径和堆栈暴露给客户端;
- 认证失败不要提示“账号存在但密码错误”之类可枚举信息;
- 对外消息保持克制,详细原因保留在受保护的日志和追踪系统中。
日志也不是安全区。请求体、文件、Token 和供应商响应在写入日志前同样需要脱敏,并且应该设置访问权限与保留周期。