本文目录
路由函数里直接 SessionLocal()、直接读全局配置、直接 new 一个邮件客户端——代码能跑,但每换一个环境(测试、预发、生产)就要改函数体,或者在一堆 if TESTING: 里打补丁。控制反转(IoC) 把「谁创建依赖、谁持有依赖」的决定从业务函数里挪出去:函数只声明「我需要什么」,具体实例由外部容器或框架在调用前塞进来。FastAPI 的 Depends 就是这条链路上最常用的一站:类型标注 + 一个小函数,框架在每次请求时按图索骥把依赖树拼好。
IoC 与依赖注入:先建立直觉
依赖是函数运行时需要的外部能力:数据库 Session、当前用户、配置对象、HTTP 客户端。传统写法在函数内部自己 new 或从全局单例取——这叫控制正转:业务代码控制创建时机和具体类型。
依赖注入(DI) 反过来:函数参数里写「给我 Session」,创建 Session 的逻辑放在别处(工厂函数、容器、框架)。好处可以拆成三点:
| 维度 | 自己 new | 注入 |
|---|---|---|
| 换实现 | 改函数体或 monkey patch | 换注入函数 / override |
| 测试 | 难隔离真实 DB | 注入内存版 / Fake |
| 可读性 | 业务与基础设施缠在一起 | 签名即文档 |
IoC 是更宽的概念:控制权反转——不限于构造函数参数,还包括「谁调用谁」。FastAPI 在请求进入路由前,先跑完依赖链,再把结果当参数传入——这就是框架层面的 IoC。
和 Java Spring 那种重量级容器不同,FastAPI 的 DI 没有运行时反射扫描整个项目;依赖图完全由类型标注和 Depends(...) 在定义路由时声明出来。这意味着:读一个 endpoint 的函数签名,就能列出它会触达哪些外部系统——对 code review 和 onboarding 很友好。代价是依赖关系分散在各路由上,大型项目会把 get_* 收到 deps.py,用 Annotated 别名(如 DbDep)避免签名过长。
Depends:声明式依赖树
Depends(callable) 告诉 FastAPI:这个参数不要从 query/body 解析,请执行 callable(可以是函数或另一个带 Depends 的可调用对象),把返回值绑到参数上。依赖可以嵌套:B 依赖 A,C 依赖 B,框架按拓扑顺序解析。
下面是一个简化的订单查询场景(与课件 OSS 上传无关):路由需要 DB Session 和「当前租户 ID」。
from collections.abc import Generator
from typing import Annotated
from fastapi import Depends, FastAPI, Header, HTTPException
from sqlalchemy.orm import Session
app = FastAPI()
def get_db() -> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db
finally:
db.close()
def get_tenant_id(x_tenant_id: Annotated[str | None, Header()] = None) -> str:
if not x_tenant_id:
raise HTTPException(status_code=400, detail="X-Tenant-Id required")
return x_tenant_id
DbDep = Annotated[Session, Depends(get_db)]
TenantDep = Annotated[str, Depends(get_tenant_id)]
@app.get("/orders/{order_id}")
def get_order(order_id: int, db: DbDep, tenant_id: TenantDep) -> dict[str, object]:
order = db.query(Order).filter(Order.id == order_id, Order.tenant_id == tenant_id).first()
if order is None:
raise HTTPException(status_code=404, detail="order not found")
return {"id": order.id, "total": order.total, "tenant_id": tenant_id}要点:
Annotated[Session, Depends(get_db)]是 Python 3.9+ 推荐写法,类型检查器认Session,FastAPI 认Depends。get_tenant_id本身也是依赖,可以读 Header、Cookie、JWT——路由函数不必碰原始请求对象。- 同一依赖函数在同一请求内默认只执行一次(缓存),多个参数都
Depends(get_db)时共享同一个 Session 实例。
子依赖与解析顺序
FastAPI 从路由函数的参数出发,递归解析每个 Depends。若 get_order_service 依赖 get_order_repo,而 get_order_repo 又依赖 get_db,则顺序一定是 先 get_db,再 repo,再 service。循环依赖(A 依赖 B、B 又依赖 A)会在启动或首次请求时报错,设计时应让依赖图保持 DAG。
Security 相关依赖(OAuth2PasswordBearer 等)在机制上与普通 Depends 相同,只是失败时倾向直接 401。可以把「解析 Token」与「加载 User 模型」拆成两个依赖:前者只返回 str | None,后者在缺失时抛 HTTPException,便于部分路由 optional 登录、部分强制登录。
APIRouter 与全局依赖
APIRouter(dependencies=[Depends(require_api_key)]) 可以把一组路由共用的门禁挂到 router 上,而不用在每个函数重复写参数。子 router include_router 时,父级 dependencies 与子级合并执行。适合管理后台整段 URL 前缀都要鉴权的场景;注意 router 级依赖不会出现在 OpenAPI 的 operation 参数列表里,文档上可能不如显式参数直观,团队要在「DRY」与「文档可读」之间取舍。
yield 依赖:进入与退出成对
上面 get_db 用 yield 而不是 return:FastAPI 把它当成生成器依赖——yield 之前是 setup(创建 Session),请求处理完后跑 finally(关闭 Session)。语义类似 with,适合连接、锁、临时目录等需要 teardown 的资源。
from contextlib import contextmanager
from collections.abc import Iterator
@contextmanager
def request_scoped_cache() -> Iterator[dict[str, object]]:
cache: dict[str, object] = {}
try:
yield cache
finally:
cache.clear()
def get_request_cache() -> Iterator[dict[str, object]]:
with request_scoped_cache() as cache:
yield cache若 yield 之后、清理之前抛错,清理仍会执行;若依赖链里某个环节失败,已成功的 yield 依赖会按相反顺序做 teardown。别把长生命周期连接池放在「每次请求 yield 一个连接」里搞混——池通常是应用级单例,yield 的是从池借出的连接。
若 yield 依赖的 setup 阶段本身抛错(例如 DB 连不上),FastAPI 不会执行该依赖的 finally 里 yield 之后的部分——因为还没 yield 成功;但已成功的外层 yield 依赖仍会 teardown。多层 yield 嵌套时,想象成栈:谁后 enter 谁先 exit。
def get_settings() -> Settings:
return settings # 应用启动时加载的单例,用 return 即可
def get_http_client(settings: Annotated[Settings, Depends(get_settings)]) -> Iterator[httpx.AsyncClient]:
client = httpx.AsyncClient(base_url=str(settings.api_base), timeout=10.0)
try:
yield client
finally:
# 关闭客户端,释放连接
import asyncio
asyncio.get_event_loop().run_until_complete(client.aclose())异步路由里更推荐 async def get_http_client() 并在 finally 里 await client.aclose(),避免在同步 generator 里硬跑 event loop。FastAPI 同时支持 sync 与 async 依赖,不要在 async 依赖里阻塞 IO。
类与可调用对象作为依赖
依赖不必是函数。带 __call__ 的类可以在 __init__ 里接收子依赖,适合有少量状态的注入点:
class RateLimiter:
def __init__(self, tenant_id: TenantDep) -> None:
self.tenant_id = tenant_id
def check(self, key: str, limit: int) -> None:
# 伪代码:按 tenant + key 查计数
...
def get_limiter(tenant_id: TenantDep) -> RateLimiter:
return RateLimiter(tenant_id)
LimiterDep = Annotated[RateLimiter, Depends(get_limiter)]FastAPI 同样支持把类直接写在 Depends(RateLimiter) 里,由框架调 RateLimiter(...) 并解析构造函数上的依赖——与显式工厂函数二选一,团队选一种风格保持一致即可。
Settings 与配置对象也适合做成依赖:get_settings() 读环境变量或 .env,测试里 override 成 Settings(database_url="sqlite:///:memory:"),避免改 os.environ 的顺序问题。配置本身无 per-request 状态时,用 return 依赖 + 模块级缓存即可,不必 yield。
为什么可测试:override 依赖
测试最怕「一 import 就连真数据库」。FastAPI 提供 app.dependency_overrides:在 TestClient 或 pytest fixture 里,把 get_db 换成返回内存 Session 或 Fake 的函数,不用改路由源码。
from fastapi.testclient import TestClient
def get_db_fake() -> Generator[Session, None, None]:
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
def test_get_order_not_found() -> None:
app.dependency_overrides[get_db] = get_db_fake
client = TestClient(app)
try:
resp = client.get(
"/orders/999",
headers={"X-Tenant-Id": "t1"},
)
assert resp.status_code == 404
finally:
app.dependency_overrides.clear()生产代码路径不变;测试只换「注入的实现」。这也是 DI 的核心 payoff:边界清晰——路由测业务编排,Repository 测 SQL,依赖注入测「换 Fake 后路由是否按预期调 Service」。
dependency_overrides 的键是原始 callable(注册时的 get_db 函数对象),不是字符串名。若把 get_db 包在 lambda 里复制多份,override 要对注册到 Depends 里的那个函数。多个测试并发跑同一 app 实例时,overrides 是全局 dict,会互相踩——应用级 fixture 里 clear(),或每个测试 create_app() 拿独立 app 实例。
对比不用 DI 的测试:往往只能 patch("myapp.routes.SessionLocal"),路径 fragile,且 patch 的是「实现细节」而非「声明的插槽」。Depends + override 测的是契约:路由声明它需要 Session,测试证明在 Fake Session 下行为正确。
与手写 Service 定位器
小团队有时在模块里放 def current_db(): return _thread_local.session,也能跑。和 Depends 比,差异在于:
| 手写定位器 | Depends |
|---|---|
| 隐式 ThreadLocal,读代码难发现 | 签名显式 |
| 测试要 patch 模块全局 | override 键清晰 |
| 难生成 OpenAPI 无关但一致的注入 | 与 FastAPI 生命周期集成 |
项目变大后,定位器往往演化成「半容器」;不如一开始把 get_* 收到 deps.py,Service 构造函数仍接受接口,路由只负责注入。
请求生命周期里 Depends 何时执行
可以把一次 HTTP 请求想成四步:匹配路由 → 解析依赖树 → 执行路由函数 → 清理 yield 依赖。第二步与第四步对调试泄漏最关键。若你在依赖里打日志,会发现 get_db 在路由函数第一行之前就已完成 setup;路由 return 之后,同一请求的 get_db finally 才 close。多个路由匹配失败不会执行你的业务依赖——只有最终命中的 endpoint 的 Depends 会跑,这与中间件「凡请求必跑」形成对比。
异步路由 async def endpoint 应配 async 依赖或同步轻量依赖;FastAPI 会在线程池跑同步阻塞依赖,避免卡 event loop,但高并发下线程池仍可能成为瓶颈。纯 CPU 或 IO 重的同步依赖,评估改成 async 实现或挪到后台任务。
与业务逻辑篇的衔接
上一篇把复杂规则从路由抽到 Service;本篇解决 Service 和 Session 怎么进路由。典型链条是:get_db → get_order_repo(db) → get_order_service(repo) → 路由参数 service: OrderServiceDep。路由不再知道 Session 从哪来,Service 不知道 HTTP Header 长什么样——租户 ID 在 get_tenant_id 依赖里解析完,以普通 str 传给 Service。这条线走通后,异常处理、中间件、测试 override 都有明确的挂载点。
安全相关依赖的拆分
鉴权常见拆法:oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") 只负责从 Header 取出 bearer 字符串;get_current_user(token: Annotated[str, Depends(oauth2_scheme)], db: DbDep) 再查库校验。这样公开路由可以不声明 get_current_user,而需要登录的路由显式加上 UserDep。Optional 用户:get_current_user_optional 在 token 缺失时返回 None 而不是 401,适合「登录后多显示一块信息」的只读接口。不要把 JWT 解码和权限 bitmask 全塞进一个巨型依赖——拆成可组合的小依赖,测试 override 其中一个即可。
并行请求与 Depends 缓存
同一时刻上千个请求各自解析 Depends,彼此独立:每个请求有自己的 Session、租户 ID、用户对象。切勿在依赖函数里用模块级可变全局存「当前用户」——并发下会串数据。若必须跨函数共享请求内状态,用 request.state 或 contextvars,生命周期与请求对齐。Depends 的「单请求缓存」指同一 coroutine 调用链里重复 Depends(get_db) 只开一次 Session,不是跨请求共享。
OpenAPI 与 Depends 的可见性
默认情况下,被 Depends 注入的参数不会全部出现在 Swagger「Parameters」里——FastAPI 知道它们不是 query/body。Security scheme 例外,会显示 Authorize 按钮。若希望文档里显式列出某 Header(如 X-Tenant-Id),应把它声明在依赖函数的参数上并使用 Header(),而不是在路由里硬读 request.headers。这对前后端联调很重要:文档即契约,缺了 tenant header 的说明,前端容易漏传而在生产才 400。
与 lifespan 初始化配合
应用启动时创建连接池、加载 ML 模型,应在 lifespan 里完成并把句柄挂在 app.state。依赖里 def get_pool(request: Request): return request.app.state.pool 把全局资源桥接成请求级可访问,而不 import 全局变量。关闭时在 lifespan finally 里 drain pool。测试可替换 lifespan 或使用 TestClient 的 with 触发 startup/shutdown,验证泄漏。
常见误解
误解 1:Depends 等于单例。 默认是「每请求、每依赖键缓存一次」,不是全局单例。要应用级单例应自己用模块级变量或 lifespan 初始化,再在依赖里返回引用。
误解 2:什么都塞进 Depends。 纯领域计算、无 IO 的工具函数不必强行 DI;值得注入的是有副作用、有实现变种、测试要替换的东西。
误解 3:yield 依赖可以省略 finally。 会话泄漏、连接占满池,往往来自「return Session 而不关闭」。资源型依赖坚持 yield + finally 或 @contextmanager。
误解 4:依赖里写重逻辑。 get_current_user 里解析 JWT、查库可以;把整段「创建订单」业务写进依赖则层次混乱——依赖提供上下文与能力,业务留在 Service 或路由调用的函数里。
小结:把 Depends 当架构文档
回顾一遍:IoC 让创建依赖的责任离开业务函数;Depends 在每次请求前拼依赖树;yield 依赖管资源 teardown;dependency_overrides 让测试换 Fake 而不改路由。落地时从 get_db、get_settings 两个依赖开始,每引入一个新外部系统(邮件、缓存、支付),先问「路由签名里是否该多一个 Depends」,而不是在 Service 里偷偷 import 全局客户端。团队 code review 看到路由里 SessionLocal() 或 requests.get 直连,就提示抽到依赖。坚持几周,测试会变简单,新同事读 endpoint 签名就能画出依赖图——这比注释更不易过期。
把 Depends 当成「函数签名上的插槽」:路由写「要什么」,测试和生产各自决定「给什么」。下一篇看错误从依赖链或业务里抛出后,如何变成长得一样的 HTTP 响应体。
依赖注入不是 FastAPI 独有,但框架把「声明 + 解析 + 清理 + override」做成一等公民,减少样板代码。对比 Flask 扩展或 Django 的 request 全局,Depends 更显式、可类型检查、可测。记住三个动词:声明(Annotated)、解析(框架)、替换(override)——掌握这三步,就掌握 FastAPI 项目里最常用的 IoC 入口。 掌握 Depends,就掌握 FastAPI 可测试性的枢纽。