FastAPI-依赖注入与异常中间件

发表于 2026-11-06 更新于 2026-11-07 分类于 Python Python教程 FastAPI 本文字数： 12k 阅读时长 ≈ 22 分钟

这篇讲什么

Depends 是 FastAPI 的核心机制：鉴权、数据库会话、分页参数、配置都写成依赖函数，处理器只声明「我要什么」。
三条实测出来的硬规矩：同一个依赖函数在一次请求里只跑一次；处理器抛异常时 yield 依赖的收尾代码不会执行；路由 404 走的是 Starlette 的 HTTPException，只注册 FastAPI 那个 handler 拦不到。
中间件、异常处理、后台任务、lifespan 四类扩展点逐个跑过，含执行顺序与 CORS 预检的响应头。

依赖的四种写法

函数依赖，可以嵌套：

deps_chain.py

from fastapi import Depends, FastAPI

app = FastAPI()


async def dep_a():
    return "A"


async def dep_b(a: str = Depends(dep_a)):
    return a + "B"


async def dep_c(a: str = Depends(dep_a)):
    return a + "C"


@app.get("/chain")
def chain(b: str = Depends(dep_b), c: str = Depends(dep_c)):
    return {"b": b, "c": c}

1 2	GET /chain -> {'b': 'AB', 'c': 'AC'} 依赖函数实际调用顺序: ['a', 'b', 'c'] # dep_a 只跑了一次

依赖树先解析、再执行，同一个可调用对象在一次请求内结果被缓存（实测两个参数共用 dep_a，计数只加 1）。
缓存是按「函数对象」算的，两个不同的函数哪怕内容一样也会各跑一次。

类依赖，参数直接变成查询参数：

class_dep.py

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class CommonQuery:
    def __init__(self, page: int = 1, size: int = 20):
        self.page, self.size = page, size


@app.get("/cls")
def cls_ep(q: Annotated[CommonQuery, Depends()]):
    return {"page": q.page, "size": q.size}

1 2	/cls -> {'page': 1, 'size': 20} /cls?page=3&size=5 -> {'page': 3, 'size': 5}

这是「一堆接口共用同一组分页参数」的标准做法，比在每个函数里写一遍干净。
Depends() 不传参数时，FastAPI 直接拿类型当依赖，所以 Annotated[CommonQuery, Depends()] 就够了。
⚠️ 别在 __init__ 里用 assert 做校验，实测 assert size <= 100 失败时 AssertionError 一路冒到服务端，客户端拿到 500；要拒绝就 raise HTTPException(422, ...)。

yield 依赖，前半段是 setup、后半段是 teardown（数据库会话、临时文件都靠它）：

yield_dep.py

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


async def dep_session():
    print("open")
    yield "sess"
    print("close")


@app.get("/use")
def use(s: Annotated[str, Depends(dep_session)]):
    return {"s": s}

收尾时机实测得说清楚，四种路径结果不一样：

/e4 正常返回      -> ['open', 'handler', 'close']
/e1 handler 抛 HTTPException(400) -> ['open', 'handler']        # close 没跑
/e2 参数校验失败(422)              -> ['open']                    # 处理器都没进，close 没跑
/e3 handler 里 1/0 (500)           -> ['open', 'handler']        # close 没跑

也就是说，裸写的 yield 依赖不保证收尾：异常路径由异常中间件短路，yield 之后的代码根本走不到。

但把收尾包进 try/finally 就完全不同，四条路径逐个实测（open = 进入依赖，close = yield 之后的收尾）：

路径	裸 `yield`	`try/finally` 包住
200 正常返回	`['open', 'handler', 'close']`	`['open', 'handler', 'close']`
400 handler 抛 `HTTPException`	`['open', 'handler']`（close 丢了）	`['open', 'handler', 'close']`
422 参数校验失败（处理器没进）	`['open']`（close 丢了）	`['open', 'close']`
500 handler 里 `1/0`	`['open', 'handler']`（close 丢了）	`['open', 'handler', 'close']`

所以规矩很明确：依赖里做资源管理就必须 try/finally，db.close()、file.close()、释放锁这类事不能指望裸 yield 后面那几行。数据库那篇的 session 依赖就是按 try/finally 写的，见 FastAPI-数据库。

依赖里可以读请求体，实测把 Pydantic 模型直接当依赖参数：

dep_body.py

from fastapi import Depends
from pydantic import BaseModel


class Payload(BaseModel):
    a: int
    b: str = "x"


async def dep_model(p: Payload):
    return p


@app.post("/dep-model")
def dep_model_ep(p: Annotated[Payload, Depends(dep_model)]):
    return {"got": p.model_dump()}

1 2	POST /dep-model {"a":1,"b":"hello"} -> 200 {"got": {"a": 1, "b": "hello"}} POST /dep-model {"b":"hello"} -> 422 missing@body.a

老教程常说「依赖里不能声明 body」，那是指同一个依赖里既想读 body 又想被多个接口复用的场景不划算；单纯读 body 实测没问题，校验错误也是标准的 422。

依赖挂在哪一层

四个层级都能挂，实测计数能看出差别：

dep_layers.py

from fastapi import APIRouter, Depends, FastAPI

app = FastAPI(dependencies=[Depends(g_dep)])        # 全局
api = APIRouter(dependencies=[Depends(r_dep)])     # router 级


@api.get("/a")
def a_ep():
    return {"ok": "a"}


@api.get("/b", dependencies=[Depends(x_dep)])      # 路由级
def b_ep():
    return {"ok": "b"}


app.include_router(api)

1	GET /a 一次 + GET /b 一次之后: {'global': 2, 'router': 2, 'route': 1}

选择原则：
- 全局依赖只放「所有接口都要」的东西，比如访问日志、请求 ID。
- 一个容易讲错的点：全局依赖不会挡文档路由。实测挂上 dependencies=[Depends(verify_admin)] 之后 /api 返回 401，而 /docs、/redoc、/openapi.json 仍然是 200。所以「给文档加登录」要单独包 docs 路由，别指望这层帮你挡。
- router 级适合「这一组接口要登录」。
- 路由级适合单个接口的权限（dependencies=[Depends(require_role("admin"))]）。
- 需要拿到依赖的返回值就用参数形式（user: User = Depends(current_user)），只需要「执行一下」就用 dependencies=[...]。
Security 是 Depends 的加强版，多一个 scopes 声明，用在 OAuth2 权限上，见 FastAPI-安全与认证。

配置对象也走依赖，但要加 lru_cache，否则每个请求都重新解析一遍环境变量与 .env：

settings_dep.py

from functools import lru_cache
from typing import Annotated

from fastapi import Depends, FastAPI
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    app_name: str = "shop"
    page_size: int = 20


@lru_cache
def get_settings():
    return Settings()


app = FastAPI()


@app.get("/cfg")
def cfg(s: Annotated[Settings, Depends(get_settings)]):
    return {"app_name": s.app_name, "page_size": s.page_size}

1
2
3

GET /cfg -> {'app_name': 'shop', 'page_size': 20}
lru_cache 命中（同一个实例）: True
不包 lru_cache 时 Settings() is Settings() -> False     # 每次都是新对象

不加 @lru_cache 就是「每个请求 new 一个 Settings」，功能没错但白花时间，还让测试里改环境变量变得难以预期。
想在测试里替换它，app.dependency_overrides[get_settings] = lambda: 假配置 正好用上（依赖是按函数对象匹配的）。

测试时替换依赖

dependency_overrides 是 FastAPI 官方推荐的「打桩」方式，实测：

override_demo.py

async def need_token(x_token: Annotated[str, Header()]):
    if x_token != "secret":
        raise HTTPException(401, "坏 token")
    return x_token


app.dependency_overrides[need_token] = lambda: "fake"

没覆盖，带坏 token -> 401
覆盖后（连头都不用带） -> {'t': 'fake'}
覆盖成返回 Response 的函数 -> 200 {"t":{"status_code":418,...}}   # 不是 418！
app.dependency_overrides.clear() 之后 -> {'t': 'secret'}

注意第三行：覆盖函数返回一个 Response 对象并不会短路请求，它只是被当成「依赖的值」塞进处理器参数再序列化出去。想模拟 401，就在覆盖函数里 raise HTTPException(401)。
覆盖是按函数对象匹配的，所以要覆盖的必须是同一个 need_token；包一层 lambda 或用子依赖会让它找不到。
记得 clear()，否则同一进程里的后续测试全被污染。

异常处理

业务错误就一个写法：raise HTTPException，别 return：

raise_only.py

@app.get("/goods/{goods_id}")
def get_goods(goods_id: int):
    if goods_id != 1:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="商品不存在")
    return {"goods_id": 1}

return HTTPException(...) 实测会把异常对象当数据序列化出去（得到一个 200 的奇怪响应），不报错、不拦截，非常难查。

三类 handler 各管一段，实测：

handlers.py

from fastapi import FastAPI, HTTPException, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()


@app.exception_handler(HTTPException)
async def h_http(request: Request, exc: HTTPException):
    return JSONResponse({"code": exc.status_code, "msg": exc.detail},
                        status_code=exc.status_code)


@app.exception_handler(RequestValidationError)
async def h_val(request: Request, exc: RequestValidationError):
    return JSONResponse({"code": 422, "msg": "参数不对", "detail": exc.errors()},
                        status_code=422)


@app.exception_handler(Exception)
async def h_any(request: Request, exc: Exception):
    return JSONResponse({"code": 500, "msg": "服务器内部错误"}, status_code=500)

/http      -> 404 {"code": 404, "msg": "自定义没有"}
/val?n=abc -> 422 {"code": 422, "msg": "参数不对", "detail": [{"type": "int_parsing", ...}]}
/crash     -> 500 {"code": 500, "msg": "服务器内部错误"}
/nope      -> 404 {"detail": "Not Found"}          # 没走自定义 handler

最后那行是重点：路由不匹配时抛的是 starlette.exceptions.HTTPException，不是 fastapi.HTTPException，所以只注册 FastAPI 那个拦不到 404。实测两种注册方式：

只注册 fastapi.HTTPException:
    /known -> 404 {"custom": "fastapi", "code": 404}      # 自己 raise 的能拦
    /nope  -> 404 {"detail": "Not Found"}                  # 路由 404 拦不到
改成注册 starlette.exceptions.HTTPException:
    /nope  -> 404 {"custom": "starlette", "code": 404, "detail": "Not Found"}

要统一全站 404 文案，就注册 Starlette 那个（fastapi.HTTPException 是它的子类，注册子类只覆盖自己 raise 的场景）。

Exception handler 有两个限制要知道：
- 它只在没有更具体的 handler 时才生效，且必须返回 Response（实测返回 dict 会 TypeError: 'dict' object is not callable）。
- 用 TestClient 验它要传 raise_server_exceptions=False，否则异常会直接冒到测试代码里。
校验错误想改文案而不是改结构，不用写 handler，改 Field(description=...) 或 Field(title=...) 就行；要改结构才动 RequestValidationError handler。

中间件

三种写法，实测执行顺序是「后加的在外层」：

middleware_order.py

from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def mw_http(request: Request, call_next):
    trace.append("http-in")
    resp = await call_next(request)
    resp.headers["X-MW"] = "http"
    trace.append("http-out")
    return resp


class MwAsgi:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        trace.append("asgi-in")
        await self.app(scope, receive, send)
        trace.append("asgi-out")


app.add_middleware(MwAsgi)

1 2	GET /mw -> ['asgi-in', 'http-in', 'handler', 'http-out', 'asgi-out'] 响应头 X-MW: http

@app.middleware("http") 是 BaseHTTPMiddleware 的糖，好处是能拿到 Response 对象改头；代价是它会把响应包成流，不适合 StreamingResponse 与 WebSocket（WebSocket 路径上它根本不参与）。
纯 ASGI 中间件（自己写 __call__）性能最好、能处理所有类型，官方文档里的 CORS/GZip 也都是纯 ASGI 实现。要埋点、计时、注入请求 ID 优先用它。

CORS 实测（CORSMiddleware）：

cors_app.py

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(CORSMiddleware,
                   allow_origins=["https://good.example"],
                   allow_credentials=True,
                   allow_methods=["GET", "POST"],
                   allow_headers=["X-Trace"],
                   expose_headers=["X-Total"])

带 Origin: https://good.example
  -> access-control-allow-origin: https://good.example
     access-control-allow-credentials: true
     access-control-expose-headers: X-Total

带 Origin: https://evil.example
  -> 没有 access-control-allow-origin（浏览器会拦），但另外两个头仍然在

OPTIONS 预检（POST + X-Trace）
  -> 200 access-control-allow-methods: GET, POST
        access-control-max-age: 600
        access-control-allow-headers: Accept, Accept-Language, Content-Language,
                                      Content-Type, X-Trace
OPTIONS 预检（DELETE）
  -> 400 Disallowed CORS method

allow_credentials=True 与 allow_origins=["*"] 是互斥的组合：带了凭据就不能用通配 origin，实测通配时浏览器会拒。要「任意源 + 凭据」得显式列名单或用 allow_origin_regex。
allow_headers 里那几个 Accept/Content-Type/... 是中间件自带的默认值，不用自己写全。
前端要读自定义响应头（比如分页的 X-Total）必须写进 expose_headers，否则 JS 拿不到。

GZip 实测：
1
2
/big (1000 个 x，minimum_size=200) -> content-encoding: gzip
/small (1 个 x) -> 没有 content-encoding
- 小于 minimum_size 不压，避免负收益。
- 生产里这层通常交给 nginx，不必让应用压；TestClient 会自动解压，所以看到的长度是解压后的，别拿它当压缩率。

后台任务

响应发出去之后再做的事，用 BackgroundTasks：

background.py

import time

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def bg_sync(name: str):
    time.sleep(0.2)
    seq.append("sync-%s" % name)


async def bg_async(name: str):
  await asyncio.sleep(0.1)
  seq.append("async-%s" % name)


@app.post("/task")
def task(bg: BackgroundTasks):
    bg.add_task(bg_sync, "1")
    bg.add_task(bg_async, "2")
    seq.append("handler")
    return {"ok": True}

1 2	POST /task -> {'ok': True} 执行顺序: ['handler', 'sync-1', 'async-2']

顺序是「先加的先跑」，而且同步任务真的会阻塞后续异步任务（sync 睡 0.2 秒，async 只能等它）。任务之间要并发就别用这个，交给任务队列（Celery / RQ / Dramatiq），写法与取舍见 Django-Celery 那类文章。
它只在「响应已经发出去」之后跑，客户端拿不到它的结果，所以不能用来保证投递：进程被杀任务就没了。发通知、写审计日志可以，扣款不行。
后台任务里抛异常，实测在 TestClient 下会直接冒到测试代码（RuntimeError）；真实服务里是记日志，响应已经发出去了，用户看不到。所以要监控就得自己包 try/except 上报。

lifespan

启动/关闭时的一次性动作，官方现在只推荐 lifespan：

lifespan_app.py

from contextlib import asynccontextmanager

from fastapi import FastAPI, Request

app = FastAPI()


@asynccontextmanager
async def lifespan(application: FastAPI):
    application.state.db = "连接池"
    print("lifespan 启动")
    yield
    print("lifespan 关闭，释放", application.state.db)


app = FastAPI(lifespan=lifespan)


@app.get("/state")
def state(request: Request):
    return {"db": request.app.state.db}

1
2
3

进入 TestClient 上下文 -> lifespan 启动
GET /state -> {'db': '连接池'}
退出上下文 -> lifespan 关闭，释放 连接池

资源挂在 app.state 上，处理器里通过 request.app.state 或 Depends 取，比全局变量好测。
必须用 with TestClient(app) as client: 才会触发，实测两种写法的事件差别：
1
2
不用 with: []
用 with: ['startup', 'shutdown']
这是测试里最常见的「为什么我的连接池没建」。

老写法 @app.on_event("startup") 实测还能用，但会打：
1
DeprecationWarning: on_event is deprecated, use lifespan event handlers instead.
而且它有个已知缺陷：startup 里 yield 不出东西，没法做「配对释放」，多资源时顺序也难保证。新代码一律 lifespan。