#FastAPI异常处理完全指南
📂 所属阶段：第二阶段 — 进阶黑科技（核心篇）
🔗 相关章节：FastAPI中间件应用 · FastAPI APIRouter模块化
#一、FastAPI异常处理基础
#1.1 异常体系结构
FastAPI 直接继承 Starlette 的异常系统，覆盖「业务预期错误」「请求验证错误」「兜底未知错误」三类核心场景：
StarletteHTTPException（基类）
├── HTTPException（业务可控异常）
│ ├── 400/401/403/404/422/自定义5xx
│ └── 可附加 headers/自定义 detail
└── RequestValidationError（Pydantic/请求自动验证错误）
Python 内置异常（ValueError/KeyError/...）
└── 未路由捕获时 → 500 Internal Server Error（需兜底）#1.2 基础的 HTTPException 使用
直接抛出 HTTPException 即可触发 FastAPI 自动响应：
from fastapi import FastAPI, HTTPException
app = FastAPI()
fruits = {"1": "Apple", "2": "Banana", "3": "Orange"}
@app.get("/fruits/{fruit_id}")
async def get_fruit(fruit_id: str):
if fruit_id not in fruits:
raise HTTPException(
status_code=404,
detail=f"Fruit {fruit_id} not found",
headers={"X-Error-Type": "FruitMissing"} # 可选附加header
)
return {"fruit_id": fruit_id, "name": fruits[fruit_id]}默认响应为简洁的 JSON：
{"detail": "Fruit 99 not found"}#二、全局异常处理器
单路由的 try/except 会导致格式混乱、敏感信息泄露，全局处理器是企业级开发的首选。
#2.1 核心处理器体系
注册以下三类处理器覆盖99%场景：
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
import logging
import traceback
from datetime import datetime
logger = logging.getLogger(__name__)
app = FastAPI()
# 1. StarletteHTTPException（含FastAPI的HTTPException）
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
return JSONResponse(
status_code=exc.status_code,
content={
"code": exc.status_code,
"message": exc.detail,
"path": str(request.url.path),
"timestamp": datetime.utcnow().isoformat()
}
)
# 2. 请求验证异常（Pydantic自动校验失败）
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
errors = []
for err in exc.errors():
field = ".".join(str(loc) for loc in err["loc"]) # 格式化错误字段路径
errors.append({"field": field, "msg": err["msg"]})
return JSONResponse(
status_code=422,
content={
"code": 422,
"message": "请求参数校验失败",
"errors": errors,
"path": str(request.url.path),
"timestamp": datetime.utcnow().isoformat()
}
)
# 3. 兜底通用异常（未路由/未预期错误）
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
# 生产环境只记录日志，不暴露traceback
logger.error(
"Unhandled Server Error",
path=str(request.url.path),
error_type=type(exc).__name__,
error_msg=str(exc),
traceback=traceback.format_exc() if app.debug else None
)
return JSONResponse(
status_code=500,
content={
"code": 500,
"message": "服务器内部错误，请稍后联系管理员",
"path": str(request.url.path),
"timestamp": datetime.utcnow().isoformat()
}
)#2.2 注意处理器注册顺序
必须先注册「具体异常」，再注册「通用异常」，否则具体处理器永远不会生效：
# ❌ 错误：通用处理器放在前面
@app.exception_handler(Exception)
async def wrong_order(...): ...
@app.exception_handler(ValueError)
async def never_triggered(...): ...
# ✅ 正确
@app.exception_handler(ValueError)
async def value_error(...): ...
@app.exception_handler(Exception)
async def general(...): ...#三、自定义业务异常类
用 HTTPException 硬编码状态码和错误信息太粗糙，统一的自定义业务异常体系是企业级应用的标配。
#3.1 定义核心异常类
创建 exceptions.py 管理所有业务异常：
from fastapi import HTTPException
from typing import Optional, Dict, Any
class AppException(HTTPException):
"""基础业务异常，封装error_code等字段"""
def __init__(
self,
status_code: int = 400,
message: str = "Bad Request",
error_code: Optional[str] = None,
headers: Optional[Dict[str, Any]] = None
):
super().__init__(
status_code=status_code,
detail=message,
headers=headers
)
self.app_code = error_code or f"E{status_code}"
class NotFoundException(AppException):
"""资源不存在"""
def __init__(self, resource: str, identifier: str):
super().__init__(
status_code=404,
message=f"{resource} '{identifier}' 不存在",
error_code="RESOURCE_NOT_FOUND"
)
class UnauthorizedException(AppException):
"""未授权"""
def __init__(self, reason: str = "请先登录"):
super().__init__(
status_code=401,
message=reason,
error_code="UNAUTHORIZED",
headers={"WWW-Authenticate": "Bearer"}
)
class RateLimitException(AppException):
"""请求限流"""
def __init__(self, retry_after: int = 60):
super().__init__(
status_code=429,
message=f"请求过于频繁，请{retry_after}秒后重试",
error_code="RATE_LIMIT_EXCEEDED"
)
self.retry_after = retry_after#3.2 注册业务异常处理器
专门注册 AppException 的处理器，保留 app_code 等扩展字段：
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
content = {
"code": exc.status_code,
"app_code": exc.app_code,
"message": exc.detail,
"path": str(request.url.path),
"timestamp": datetime.utcnow().isoformat()
}
response = JSONResponse(status_code=exc.status_code, content=content)
if isinstance(exc, RateLimitException):
response.headers["Retry-After"] = str(exc.retry_after)
return response#3.3 在业务逻辑中使用
from exceptions import NotFoundException, UnauthorizedException
@app.delete("/articles/{article_id}")
async def delete_article(article_id: int, current_user: dict = Depends(get_current_user)):
# 模拟数据库查询
article = {"id": article_id, "author_id": 1}
if not article:
raise NotFoundException("文章", str(article_id))
if article["author_id"] != current_user["id"]:
raise UnauthorizedException("无权删除他人文章")
return {"deleted": article_id, "message": "删除成功"}#四、统一响应格式
不仅是错误响应，成功响应也应该统一格式，方便前端统一处理。
#4.1 封装统一响应模型
创建 schemas.py 用 Pydantic 定义响应规范：
from pydantic import BaseModel, Field
from typing import TypeVar, Generic, Optional
from datetime import datetime
T = TypeVar("T")
class ApiResponse(BaseModel, Generic[T]):
"""通用成功响应"""
code: int = 200
message: str = "success"
data: Optional[T] = None
timestamp: str = Field(default_factory=lambda: datetime.utcnow().isoformat())
class ApiError(BaseModel):
"""通用错误响应"""
code: int
app_code: str
message: str
path: str
timestamp: str = Field(default_factory=lambda: datetime.utcnow().isoformat())
# 辅助函数
def success_response(data: Optional[T] = None, message: str = "success") -> ApiResponse[T]:
return ApiResponse(data=data, message=message)#五、常见陷阱与避坑
#❌ 陷阱1：暴露敏感信息
生产环境绝对不要返回 traceback 或数据库查询语句：
# ❌ 错误
@app.exception_handler(Exception)
async def leaky_handler(...):
return JSONResponse(500, {"traceback": traceback.format_exc()})
# ✅ 正确
@app.exception_handler(Exception)
async def safe_handler(...):
if app.debug:
content["traceback"] = traceback.format_exc()
return JSONResponse(500, content)#❌ 陷阱2：用异常控制正常流程
异常的创建和捕获开销较大，不要用于正常分支判断：
# ❌ 错误
def get_fruit_safe(fruit_id: str):
try:
return fruits[fruit_id]
except KeyError:
return None
# ✅ 正确
def get_fruit_optimized(fruit_id: str):
return fruits.get(fruit_id)#六、总结
| 异常类型 | 触发场景 | 处理方式 |
|---|---|---|
HTTPException/AppException | 业务可控错误 | 自定义异常类 + 统一响应格式 |
RequestValidationError | Pydantic/请求参数自动校验失败 | 格式化错误字段路径 + 统一提示 |
Exception | 所有未预期错误 | 生产环境隐藏细节 + 详细日志记录 |
💡 核心原则：客户端收到可预期的响应，服务端记录可追溯的日志。