Chapter 06 · 错误处理与日志：事故的善后
📚 学完本章，你将能够：
- 区分业务错误（如"库存不足"）和系统错误（如"数据库断开"）
- 定义统一的错误响应格式，让前端处理错误更轻松
- 用全局异常处理器捕获未预料的错误，防止泄露敏感信息
- 配置结构化日志，方便事后排查问题
- 用中间件给每个请求加上唯一 ID，实现全链路追踪
💡 场景引入
你的图书 API 上线第二天，用户反馈"有时候删除书会报错"。你去看服务器日志，发现有的错误返回了 Python 的堆栈跟踪，暴露了文件路径；有的错误干脆没记录，无从追查。
错误处理不是"锦上添花"，而是生产系统的生命线。好的错误处理让前端知道发生了什么，让开发者能定位问题，让黑客看不到内部信息。
6.1 两种错误：业务错误 vs 系统错误
| 类型 | 例子 | 谁的责任 | 返回什么 |
|---|---|---|---|
| 业务错误 | 库存不足、用户已存在、余额不够 | 用户的输入或状态 | 400 / 409 + 清晰的中文错误信息 |
| 系统错误 | 数据库断开、磁盘满、代码 bug | 服务端 | 500 + "服务器内部错误"（不暴露细节） |
6.2 不要直接抛 HTTPException
初学者喜欢在业务代码里到处写 raise HTTPException。这会导致两个问题：
- 业务逻辑和 HTTP 协议耦合——你的 Service 层不应该知道什么是 404
- 错误信息散落在代码各处，难以统一管理
❌ 业务层直接抛 HTTP 异常
async def create_order(data: OrderCreate, db: AsyncSession):
book = await db.get(Book, data.book_id)
if not book:
raise HTTPException(404, "Book not found") # 业务层知道 HTTP？
if book.stock < data.quantity:
raise HTTPException(400, "Out of stock") # 到处散落✅ 定义业务异常，全局映射
class NotFoundError(Exception):
def __init__(self, resource: str):
self.resource = resource
class OutOfStockError(Exception):
pass
async def create_order(data: OrderCreate, db: AsyncSession):
book = await db.get(Book, data.book_id)
if not book:
raise NotFoundError("Book")
if book.stock < data.quantity:
raise OutOfStockError()6.3 全局异常处理器
FastAPI 允许你注册全局异常处理器，统一转换异常为 HTTP 响应：
异常处理
from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
class ErrorResponse(BaseModel):
code: str
message: str
detail: dict | None = None
@app.exception_handler(NotFoundError)
async def not_found_handler(request: Request, exc: NotFoundError):
return JSONResponse(
status_code=404,
content=ErrorResponse(
code="not_found",
message=f"{exc.resource} not found"
).model_dump(),
)
@app.exception_handler(OutOfStockError)
async def out_of_stock_handler(request: Request, exc: OutOfStockError):
return JSONResponse(
status_code=409,
content=ErrorResponse(code="out_of_stock", message="库存不足").model_dump(),
)
@app.exception_handler(Exception)
async def catch_all_handler(request: Request, exc: Exception):
logger.exception("Unhandled error") # 记录完整堆栈
return JSONResponse(
status_code=500,
content=ErrorResponse(
code="internal_error",
message="服务器内部错误，请稍后重试"
).model_dump(),
)最后一个 @app.exception_handler(Exception) 是安全网。任何未被前面处理器捕获的异常（比如代码 bug）都会走到这里，返回统一的 500 错误，同时记录详细日志供开发者排查。
6.4 请求追踪：给每个请求一个身份证
当用户报告"下单报错"时，如果日志里有上千条记录，你怎么知道哪条是他的？给每个请求分配一个唯一 ID（request_id），并在所有日志中携带它。
中间件生成 request_id
import uuid
import time
@app.middleware("http")
async def add_request_id(request: Request, call_next):
request_id = str(uuid.uuid4())[:8]
request.state.request_id = request_id
start = time.time()
response = await call_next(request)
duration = (time.time() - start) * 1000
logger.info(
f"{request.method} {request.url.path} "
f"{response.status_code} {duration:.1f}ms req={request_id}"
)
response.headers["X-Request-ID"] = request_id
return response现在每条日志都带有 req=abc12345。用户报问题时，你把他的 request_id 一搜，所有相关日志就串起来了。
6.5 结构化日志
生产环境不要用 print()。推荐使用 loguru，配置 JSON 输出：
日志配置
from loguru import logger
import sys
logger.remove() # 移除默认 handler
# 开发环境：彩色文本
logger.add(sys.stdout, colorize=True, level="DEBUG")
# 生产环境：JSON 格式，方便 ELK/Loki 收集
logger.add(
"app.log",
serialize=True, # 输出 JSON
rotation="100 MB", # 文件超过 100MB 自动切分
retention="30 days", # 保留 30 天
level="INFO",
)🎯 知识检测
问题 1
为什么业务代码不应该直接抛
HTTPException？
解析：业务层只关心"发生了什么业务问题"（如库存不足），不应该知道 HTTP 状态码。全局异常处理器负责把业务异常翻译成 HTTP 响应。这样业务代码更纯粹，错误格式也更统一。
问题 2
全局
Exception 处理器的作用是什么？
解析：它是一张安全网。代码中总有你没考虑到的边界情况，全局处理器确保这些异常不会直接把 Python 堆栈暴露给用户（可能泄露文件路径、库版本等敏感信息），而是返回统一的"服务器错误"，同时记录日志供开发者排查。
问题 3
request_id 的主要用途是什么？
解析：高并发系统中，日志是交错打印的。request_id 就像给一次请求的全过程贴上了同一个标签，事后可以通过这个标签快速筛选出相关日志，定位问题。
本章小结
✓
业务错误 ≠ 系统错误
业务错误给用户看（库存不足），系统错误给开发者看（数据库断开）。
✓
全局异常处理器是安全网
统一错误格式，防止泄露敏感信息，记录完整堆栈。
✓
请求 ID 实现全链路追踪
给每个请求发身份证，日志里一搜就知道来龙去脉。
✓
生产日志用 JSON 结构化
方便被日志收集系统解析、检索和告警。