创建时间: 2026-08-25最后更新: 2026-08-27

1. 失败

一次接口调用失败,原因不一定都在同一层。把所有错误都返回成同一个状态码或同一句文本,会让客户端无法判断应该修改请求、重新登录、稍后重试,还是联系服务维护者。

可以先把失败分为三类:

类型示例主要处理方
请求错误缺少字段、类型错误、认证信息无效客户端修改请求或身份信息
业务错误会话不存在、余额不足、状态冲突客户端按业务状态处理
系统错误代码异常、数据库断开、上游不可用服务端记录、告警和修复

这三类失败最终都可以表现为 HTTP 响应,但处理策略不同。HTTP 状态码描述协议层结果,业务错误代码供程序稳定判断,日志和 Request ID 则帮助服务端定位问题。它们各自解决不同问题,不能互相替代。

2. HTTPException

对于已经预料到、并且能够明确转换成 HTTP 响应的失败,可以抛出 HTTPException:

http_exception.py
01
from fastapi import FastAPI, HTTPException, status
02
03
app = FastAPI()
04
05
conversations = {"conversation-1": {"title": "FastAPI"}}
06
07
@app.get("/conversations/{conversation_id}")
08
async def get_conversation(conversation_id: str):
09
conversation = conversations.get(conversation_id)
10
11
if conversation is None:
12
raise HTTPException(
13
status_code=status.HTTP_404_NOT_FOUND,
14
detail="会话不存在",
15
)
16
17
return conversation

这里必须使用 raise,而不是 return。异常抛出后,当前调用链停止执行,FastAPI 再把它转换成 HTTP 响应。因此,路由函数调用的依赖或普通工具函数也可以抛出 HTTPException,不需要层层返回错误标记。

默认响应如下:

not_found.json
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 边界统一转换。

domain_errors.py
1
class ConversationNotFoundError(Exception):
2
def __init__(self, conversation_id: str):
3
self.conversation_id = conversation_id
4
super().__init__(f"conversation not found: {conversation_id}")

业务服务只表达发生了什么:

conversation_service.py
1
from domain_errors import ConversationNotFoundError
2
3
async def find_conversation(conversation_id: str):
4
conversation = await repository.find_by_id(conversation_id)
5
6
if conversation is None:
7
raise ConversationNotFoundError(conversation_id)
8
9
return conversation

FastAPI 应用负责把它翻译成 HTTP 响应:

exception_handlers.py
01
from fastapi import FastAPI, Request, status
02
from fastapi.responses import JSONResponse
03
04
from domain_errors import ConversationNotFoundError
05
06
app = FastAPI()
07
08
@app.exception_handler(ConversationNotFoundError)
09
async def handle_conversation_not_found(
10
request: Request,
11
error: ConversationNotFoundError,
12
) -> JSONResponse:
13
request_id = getattr(request.state, "request_id", None)
14
15
return JSONResponse(
16
status_code=status.HTTP_404_NOT_FOUND,
17
content={
18
"error": {
19
"code": "conversation_not_found",
20
"message": "会话不存在",
21
"request_id": request_id,
22
}
23
},
24
)

这样,业务服务只关心“没有找到会话”,HTTP 层负责决定它对应 404 和怎样的 JSON 结构。如果同一套业务逻辑以后被命令行程序、定时任务或消息消费者调用,领域异常仍然可以复用。

实际项目中可以按业务模块定义少量有明确语义的异常,例如 ConversationNotFoundError、PermissionDeniedError 和 StateConflictError。不要为每一句错误文案都创建一个异常类,否则异常体系会比业务本身更难维护。

5. 错误协议

接口只返回自然语言时,前端只能匹配文案。更稳定的错误协议应该同时包含机器可读代码和面向人的说明:

error_response.json
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:某些规则附带的校验上下文。

如果项目需要统一外层协议,可以注册自己的处理器。下面只公开字段位置、类型和说明,没有把原始输入直接返回给客户端:

validation_handler.py
01
from fastapi import FastAPI, Request, status
02
from fastapi.exceptions import RequestValidationError
03
from fastapi.responses import JSONResponse
04
05
app = FastAPI()
06
07
@app.exception_handler(RequestValidationError)
08
async def handle_validation_error(
09
request: Request,
10
error: RequestValidationError,
11
) -> JSONResponse:
12
request_id = getattr(request.state, "request_id", None)
13
fields = [
14
{
15
"location": list(item["loc"]),
16
"type": item["type"],
17
"message": item["msg"],
18
}
19
for item in error.errors()
20
]
21
22
return JSONResponse(
23
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
24
content={
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 的类型。

http_error_handler.py
01
from fastapi import FastAPI, Request
02
from fastapi.encoders import jsonable_encoder
03
from fastapi.responses import JSONResponse
04
from starlette.exceptions import HTTPException as StarletteHTTPException
05
06
app = FastAPI()
07
08
@app.exception_handler(StarletteHTTPException)
09
async def handle_http_error(
10
request: Request,
11
error: StarletteHTTPException,
12
) -> JSONResponse:
13
request_id = getattr(request.state, "request_id", None)
14
15
if isinstance(error.detail, str):
16
content = {
17
"error": {
18
"code": f"http_{error.status_code}",
19
"message": error.detail,
20
"request_id": request_id,
21
}
22
}
23
else:
24
content = {
25
"error": {
26
"code": f"http_{error.status_code}",
27
"message": "请求失败",
28
"request_id": request_id,
29
"details": jsonable_encoder(error.detail),
30
}
31
}
32
33
return JSONResponse(
34
status_code=error.status_code,
35
headers=error.headers,
36
content=content,
37
)

业务代码仍然可以继续抛出 fastapi.HTTPException。统一格式时有两个细节不能遗漏:

  • 保留 error.headers,否则 WWW-Authenticate、Retry-After 等协议字段会丢失;
  • 保留结构化 detail 的数据结构,不能使用 str(error.detail) 把字典变成不可解析的文本。

如果只是想记录 HTTP 错误而不改变默认响应,也可以调用 FastAPI 内置的 http_exception_handler。复用官方处理器比手动复制默认 JSON 格式更稳妥。

9. 未知异常

未预期异常通常应该返回 500,但不能把堆栈、密钥、数据库语句或供应商原始响应发给客户端。可以注册一个最终兜底处理器,对外返回稳定协议,对内保留完整堆栈:

unexpected_error_handler.py
01
import logging
02
03
from fastapi import FastAPI, Request, status
04
from fastapi.responses import JSONResponse
05
06
logger = logging.getLogger(__name__)
07
app = FastAPI()
08
09
@app.exception_handler(Exception)
10
async def handle_unexpected_error(
11
request: Request,
12
error: Exception,
13
) -> JSONResponse:
14
request_id = getattr(request.state, "request_id", None)
15
16
logger.exception(
17
"Unhandled request error",
18
extra={
19
"request_id": request_id,
20
"method": request.method,
21
"path": request.url.path,
22
},
23
)
24
25
return JSONResponse(
26
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
27
content={
28
"error": {
29
"code": "internal_server_error",
30
"message": "服务暂时无法完成请求",
31
"request_id": request_id,
32
}
33
},
34
)

这里的 logger.exception() 会记录异常堆栈,公开响应只提供通用信息。生产日志通常还需要记录路由、请求方法、上游名称和耗时,但不要默认记录 Authorization、Cookie、完整请求体、提示词或文件内容。

不要在每个路由里使用下面这种宽泛捕获:

bad_example.py
1
try:
2
return await service.run()
3
except Exception:
4
return {"error": "failed"}

它不仅把 HTTP 状态码错误地变成了 200,还丢失了异常堆栈、统一协议和监控信号。只有当当前层真正知道如何恢复、补偿或翻译异常时,才应该捕获它。

开发环境中的 debug=True 会让 500 响应展示调试信息,并绕过自定义 500 页面。它只适合本地调试,生产环境必须关闭。

10. 上游错误

模型服务、数据库代理和外部 HTTP API 会产生超时、限流、认证失败、连接失败和响应协议错误。应用应该在上游适配层识别这些异常,再转换成本系统的稳定语义。

upstream_timeout.py
01
import asyncio
02
03
from fastapi import HTTPException, status
04
05
async def invoke_with_timeout(model, messages):
06
try:
07
async with asyncio.timeout(30):
08
return await model.ainvoke(messages)
09
except TimeoutError as error:
10
raise HTTPException(
11
status_code=status.HTTP_504_GATEWAY_TIMEOUT,
12
detail={
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。取消表示“这项工作已经没有继续执行的必要”,通常应该尽快结束,并把取消信号继续向上传播。

cancellation.py
1
import asyncio
2
3
async def run_model(model, messages):
4
try:
5
return await model.ainvoke(messages)
6
except asyncio.CancelledError:
7
await release_temporary_resources()
8
raise

在当前 Python 中,CancelledError 继承自 BaseException,普通的 except Exception 不会捕获它。如果为了清理资源而显式捕获,就必须在清理后再次 raise。把取消转换成 500、504 或一个空结果,会让服务继续占用已经不再需要的资源,也会破坏 TaskGroup、超时和服务关闭流程。

12. 响应时机

异常只有在响应尚未发送时,才能转换成新的 HTTP 状态码和 JSON 正文。流式响应发送第一块数据后再出错,响应头和状态码通常已经确定,此时不能把 200 改成 500。应用只能按约定发送流内错误事件、结束连接并记录日志。

后台任务也发生在响应之后。即使异常处理器收到了后台任务抛出的异常,它生成的新响应也已经无法替换先前发出的响应。后台任务必须自行记录失败、执行有限重试,或把状态写入可查询的存储。多个后台任务按顺序运行时,前一个任务抛出异常还会阻止后续任务执行。

WebSocket 升级完成后也不再使用普通 HTTP 状态码表达错误。连接建立前可以拒绝升级;连接建立后应使用 WebSocketException 或合适的关闭码结束连接。

因此,“异常能否改写 HTTP 响应”取决于错误发生的时间和协议阶段,而不只取决于异常类型。

13. 文档与测试

运行时注册了异常处理器,不代表 OpenAPI 会自动知道每个路由可能返回哪些业务错误。需要通过路径装饰器的 responses 参数声明错误模型:

documented_errors.py
01
from uuid import UUID
02
03
from fastapi import APIRouter
04
from pydantic import BaseModel
05
06
router = APIRouter()
07
08
class ErrorBody(BaseModel):
09
code: str
10
message: str
11
request_id: str | None = None
12
13
class ErrorResponse(BaseModel):
14
error: ErrorBody
15
16
@router.get(
17
"/conversations/{conversation_id}",
18
responses={
19
404: {
20
"model": ErrorResponse,
21
"description": "会话不存在",
22
}
23
},
24
)
25
async def get_conversation(conversation_id: UUID):
26
return await service.find_conversation(str(conversation_id))

responses 负责生成文档,不会替你执行异常转换,也不会在运行时强制校验手动创建的 JSONResponse。实际的异常处理器、错误协议和 OpenAPI 声明必须保持一致。多个路由共用同一组错误时,可以抽取公共 responses 字典再合并使用。

错误分支也需要测试,而且应该断言稳定字段:

test_errors.py
01
from fastapi.testclient import TestClient
02
03
from main import app
04
05
client = TestClient(app)
06
07
def test_conversation_not_found():
08
response = client.get(
09
"/conversations/123e4567-e89b-12d3-a456-426614174000"
10
)
11
12
assert response.status_code == 404
13
assert response.json()["error"]["code"] == "conversation_not_found"
14
15
def test_invalid_conversation_id():
16
response = client.get("/conversations/not-a-uuid")
17
18
assert response.status_code == 422
19
assert response.json()["error"]["code"] == "validation_error"

测试不必断言整段自然语言,因为文案可能调整。更值得验证的是状态码、错误代码、必要响应头和字段位置。测试未知 500 处理器时,可以创建 TestClient(app, raise_server_exceptions=False),否则测试客户端默认会把服务端异常重新抛给测试进程。

14. 安全

错误信息也是接口的数据边界:

  • 不返回 API Key、Authorization Header 和 Cookie;
  • 不返回完整提示词、用户隐私和模型原始调试对象;
  • 不把数据库语句、内部路径和堆栈暴露给客户端;
  • 认证失败不要提示“账号存在但密码错误”之类可枚举信息;
  • 对外消息保持克制,详细原因保留在受保护的日志和追踪系统中。

日志也不是安全区。请求体、文件、Token 和供应商响应在写入日志前同样需要脱敏,并且应该设置访问权限与保留周期。

正在验证登录状态
请稍候,验证完成后将继续显示文章内容