08 异常处理 —— 让错误响应像正常响应一样有契约

前置阅读: 04-路由、05-参数分类、06-请求与响应
关键词: HTTPException, exception_handler, IntegrityError, async def
难度: ★★★★☆

场景导入：每个 handler 里写 try/except，不累吗？

删除图书时，资源可能不存在、可能已被借出、可能外键约束阻止了操作。新增图书时，ISBN 可能重复、作者 ID 可能无效、数据库连接可能刚好断了。如果每个 handler 都自己 try/except 然后拼接错误 JSON，不出三天状态码和错误格式就会各写各的——前端对接时痛不欲生。

FastAPI 用一套统一的异常处理机制把"什么时候抛异常"和"异常怎么变成 HTTP 响应"彻底分开。handler 里只需要 raise HTTPException 或让 SQLAlchemy 的 IntegrityError 自然冒泡，全局的异常处理器负责把各种异常类型映射为一致的 JSON 结构。本篇以删除和新增两个场景为例，建立一条从异常到响应的完整链路。

原理解析：异常的三层分派

FastAPI 的异常处理建立在 Starlette 的 ExceptionMiddleware 之上。请求处理过程中，任何未捕获的异常都会沿调用栈向上冒泡，最终被这个中间件拦截。拦截后，框架按照异常的 MRO（方法解析顺序）链查找匹配的处理器——从最具体的异常类型一直找到 Exception 兜底。

注册异常处理器有两种方式，最终效果完全等价：

两种可以混用，FastAPI 在启动期会把装饰器注册的处理器合并到 exception_handlers 字典中。

处理器函数本身既可以是 def 也可以是 async def——Starlette 在调度时通过 inspect.iscoroutinefunction 判断：协程就 await，普通函数就直接调用。这个设计和 handler 的 async def/def 双形态一脉相承：只在处理器内部需要 await（如回滚事务、发送告警）时才用 async def，纯 JSON 构造直接用 def。

下面这张图把异常分派的完整逻辑画了出来：

（图注：装饰器和字典注册的处理器最终合并到同一个分派链；def 和 async def 都是合法签名。）

异步栈下一个特别重要的场景是：SQLAlchemy 在 await db.commit() 时抛出的 IntegrityError。这是 aiomysql 底层 MySQL 错误的包装——真正的错误码藏在 exc.orig.args[0] 里。常见映射关系如下：

在全局处理器里完成这些映射后，DAO 层就不需要为每个 await db.commit() 写 try/except——让异常自然冒泡，由框架统一收敛。这是"自下而上"错误处理的核心思路。

（图注：IntegrityError 在 await db.commit() 处抛出，处理器通过 exc.orig.args[0] 解析 MySQL 错误码，映射为 409。）

代码实现：三种处理器，两种形态

下面的代码在 main.py 中集中注册三个处理器。http_exception_handler 只构造响应，用 def；integrity_exception_handler 为将来可能需要的 await 操作预留了 async def 签名；兜底处理器同样用 async def。所有处理器通过 FastAPI(exception_handlers={...}) 字典形参集中管理。

# app/main.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from sqlalchemy.exc import IntegrityError


def http_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    """纯响应构造，使用 def。Starlette 自动识别为同步处理器。"""
    status_code = getattr(exc, "status_code", 500)
    detail = getattr(exc, "detail", "internal error")
    return JSONResponse(status_code=status_code, content={"detail": detail})


async def integrity_exception_handler(
    request: Request, exc: IntegrityError
) -> JSONResponse:
    """捕获 SQLAlchemy IntegrityError，异步栈下允许 await 后续动作。

    MySQL 错误码映射：
      - 1062 重复键 → 409 Conflict
      - 1452 外键约束 → 422 Unprocessable Entity
      - 1213 死锁 → 409 Conflict（可重试）
    """
    code = exc.orig.args[0] if exc.orig and exc.orig.args else None
    if code == 1062:
        return JSONResponse(status_code=409, content={"detail": "duplicate key"})
    if code == 1452:
        return JSONResponse(status_code=422, content={"detail": "fk violation"})
    if code == 1213:
        return JSONResponse(status_code=409, content={"detail": "deadlock, retry"})
    return JSONResponse(status_code=500, content={"detail": "db error"})


async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    """兜底处理器：未知异常 → 500。"""
    return JSONResponse(status_code=500, content={"detail": "internal server error"})


app = FastAPI(
    title="图书管理 API",
    version="0.3.0",
    exception_handlers={
        Exception: unhandled_exception_handler,
        IntegrityError: integrity_exception_handler,
    },
)

几处设计决策值得展开：

• Exception 兜底处理器放在字典里注册，它能拦截所有未匹配到更具体处理器的异常。但 HTTPException 本身是 Exception 的子类——如果不希望它被兜底处理器吞掉，需要额外注册 HTTPException: http_exception_handler，或让兜底处理器判断 hasattr(exc, 'status_code') 来透传。

• integrity_exception_handler 用 async def，即使当前实现没有 await。这是为将来扩展留空间——比如在处理器里 await 一个告警通知或审计日志写入。如果确认永远不需要异步操作，可以改成 def。

• 处理器签名固定为 (request: Request, exc: ExceptionType) -> Response。返回值必须是 Starlette 的 Response 对象，不能是 Pydantic 模型或字典——因为异常处理器绕过了 response_model 管线。

classDiagram
    class FastAPI {
      +exception_handlers: dict
      +add_exception_handler()
    }
    class http_exception_handler {
      <<def>>
      +__call__(request, exc)
    }
    class integrity_exception_handler {
      <<async def>>
      +__call__(request, exc)
    }
    class unhandled_exception_handler {
      <<async def>>
      +__call__(request, exc)
    }
    FastAPI --> http_exception_handler : HTTPException
    FastAPI --> integrity_exception_handler : IntegrityError
    FastAPI --> unhandled_exception_handler : Exception 兜底

（图注：def 与 async def 处理器形态并列，FastAPI 通过 iscoroutinefunction 判断调度路径。）

避坑指南

• IntegrityError 的错误码在 exc.orig.args[0]，不是 exc.code。直接 str(exc) 只能拿到 SQLAlchemy 的英文模板字符串，做不了精确分支。写处理器之前先 print(exc.orig.args) 确认实际的错误码位置。

• 错误处理器里不要再访问可能失败的数据库。错误路径只应使用内存中已有的 exc 字段来构造响应。如果在错误处理器里再发一次数据库查询，失败了会覆盖原始异常，让你排查时完全摸不着头脑。

• Response 的 status_code 和 detail 都可以通过 HTTPException 构造器传入。handler 里写 raise HTTPException(status_code=404, detail="book not found")，处理器里就能通过 exc.status_code 和 exc.detail 读取——不需要每种异常定义一个子类。

面试 QA

Q1 [原理]: @app.exception_handler 装饰器和 FastAPI(exception_handlers={...}) 字典形参有什么区别？

没有本质区别。两者最终都注册到 Starlette 的 ExceptionMiddleware，通过 MRO 匹配最优处理器。装饰器形态适合把处理器和自定义异常类放在同一模块，字典形参适合在 main.py 集中管理，让人一眼看清全局覆盖关系和兜底顺序。两种可以混用，FastAPI 启动期自动合并。

处理器函数本身 def 和 async def 都可以——框架在调度时通过 inspect.iscoroutinefunction 选择执行路径。只在处理器内需要 await（如发告警、写审计日志）时才用 async def。

Q2 [项目]: aiomysql 异步栈下常见的 SQL 错误如何映射到 HTTP 状态码？

在 main.py 集中注册一个 IntegrityError 处理器，通过 exc.orig.args[0] 拿到 MySQL 错误码后分支：

• 1062（唯一键重复）→ 409 Conflict

• 1452（外键约束失败）→ 422 Unprocessable Entity

• 1213（死锁）→ 409 Conflict，可提示客户端重试

• 其余 → 500 Internal Server Error

这样做的好处是：DAO 层不需要为每个 await db.commit() 都包一层 try/except IntegrityError。让异常沿异步栈自然冒泡，由全局处理器统一收敛为 JSON 响应——代码更简洁，错误格式更一致。

小结

本篇把异常处理从"每个 handler 自己拼错误字典"提升到了"全局注册、按类型分派"的工程化水平。核心思路就两条：handler 和 DAO 只负责 raise，处理器负责把异常翻译为一致的 JSON 结构；def 和 async def 两种处理器形态可以共存，框架自动分流。

下一篇《09 中间件》将把横切关注点从异常处理扩展到请求处理链的更外层——耗时统计、请求 ID 注入、响应头统一管理，都是中间件的拿手好戏。