FastAPI依赖注入系统设计剖析
引言
去年我接手了一个日活百万的AI服务网关，代码基于FastAPI构建。上线三个月后，一个诡异的线上问题让我彻夜难眠：在QPS突破8000时，部分请求的数据库连接竟然串了——A用户的请求拿到了B用户的租户上下文，导致跨租户数据泄露。
排查了两天，最终定位到问题根源：一个使用了Depends的依赖函数内部持有可变状态，在高并发下被多个请求共享。这个坑让我意识到，很多开发者（包括当时的我）对FastAPI依赖注入系统的理解，还停留在"写个函数加个Depends就完事"的层面，对其底层的缓存机制、作用域、生命周期一无所知。
FastAPI的依赖注入（Dependency Injection，DI）系统是整个框架最精华也最容易被低估的部分。它不像Spring那样通过注解和反射在启动时构建Bean容器，而是走了一条基于函数签名的运行时解析路线——轻量、灵活，但也暗藏陷阱。
这篇文章，我会带你从源码层面彻底搞懂FastAPI的DI系统：它如何解析依赖树？缓存是怎么工作的？yield依赖的生命周期边界在哪？以及如何在生产环境中避免我踩过的那些坑。
核心概念：从餐厅后厨说起
先打个比方。想象你经营一家餐厅，顾客点了一道"宫保鸡丁"：
- 传统方式（不用DI）：厨师做菜时，需要自己跑去仓库拿鸡肉、去菜地摘辣椒、去调料架找酱油。每个厨师都要重复这些动作，仓库和菜地还可能被同时争抢。
- 依赖注入方式：厨师只负责炒菜，食材由"备菜员"统一准备好放在灶台边。厨师说"我需要鸡肉和辣椒"，备菜员就送过来；如果两个厨师都需要同一批鸡肉，备菜员只准备一次，共享给他们。
FastAPI的DI系统就是这位"备菜员"。它的核心工作流程是：
- 声明依赖：通过
Depends(callable)在函数参数上声明"我需要什么" - 解析依赖树：运行时分析函数签名，递归构建依赖关系图
- 执行与缓存：按拓扑顺序执行依赖函数，同一请求内默认缓存结果
- 注入参数：把解析结果作为参数传给目标路由函数
用技术语言定义：FastAPI的DI是一个基于Python函数签名内省的、请求级别的依赖解析与缓存系统。它支持同步/异步混合、支持嵌套依赖、支持生成器（yield）形式的资源管理。
依赖树的形态
注意图中 get_settings 被三个依赖同时需要，get_db_session 被两个依赖需要。如果没有缓存，同一个请求内这些函数会被重复执行多次——这就是FastAPI DI缓存机制存在的意义，也是很多坑的来源。
源码深度分析
要真正理解FastAPI的DI，必须读三个文件：
fastapi/dependencies/utils.py：依赖解析的核心
fastapi/routing.py：请求处理时如何触发依赖求解
fastapi/params.py：Depends类的定义
Depends 的极简本质
先看Depends的定义（fastapi/params.py）：
class Depends:
def __init__(
self,
dependency: Optional[Callable[..., Any]] = None,
*,
use_cache: bool = True,
scope: Optional[Literal["function", "request"]] = None,
) -> None:
self.dependency = dependency
self.use_cache = use_cache
self.scope = scope
def __repr__(self) -> str:
attr = getattr(self.dependency, "__name__", type(self.dependency).__name__)
cache = "" if self.use_cache else ", use_cache=False"
return f"{type(self).__name__}({attr}{cache})"是不是简单到令人失望？Depends只是一个数据载体（marker），它本身不做任何解析工作。真正的魔法发生在 get_dependant() 和 solve_dependencies() 这两个函数里。
这是一个非常关键的设计哲学：声明与执行分离。Depends只是声明"这里有依赖"，具体的解析逻辑在路由构建阶段和请求处理阶段分别完成。
依赖树构建：get_dependant()
当一个路由被注册时，FastAPI会对路由函数调用 get_dependant()，把它的签名"翻译"成一棵 Dependant 对象树。
# fastapi/dependencies/utils.py (简化版)
def get_dependant(
*,
path: str,
call: Callable[..., Any],
name: Optional[str] = None,
security_scopes: Optional[List[str]] = None,
use_cache: bool = True,
) -> Dependant:
path_param_names = get_path_param_names(path)
endpoint_signature = get_typed_signature(call)
signature_params = endpoint_signature.parameters
dependant = Dependant(
call=call,
name=name,
path=path,
security_scopes=security_scopes,
use_cache=use_cache,
)
for param_name, param in signature_params.items():
# 递归解析：如果参数默认值是 Depends 实例
if isinstance(param.default, params.Depends):
sub_dependant = get_param_sub_dependant(
param=param, path=path, security_scopes=security_scopes
)
dependant.dependencies.append(sub_dependant)
continue
# ... 处理 Path/Query/Body 等其他参数
return dependant这里有个递归：get_param_sub_dependant 会再次调用 get_dependant，从而把嵌套依赖也构建成树。注意，这个构建过程只在应用启动时发生一次，请求处理时直接复用这棵树。
Dependant 对象长这样：
@dataclass
class Dependant:
path_params: List[ModelField] = field(default_factory=list)
query_params: List[ModelField] = field(default_factory=list)
header_params: List[ModelField] = field(default_factory=list)
cookie_params: List[ModelField] = field(default_factory=list)
body_params: List[ModelField] = field(default_factory=list)
dependencies: List["Dependant"] = field(default_factory=list)
name: Optional[str] = None
call: Optional[Callable[..., Any]] = None
use_cache: bool = True
cache_key: Optional[Tuple[str, ...]] = Nonecache_key 是理解缓存机制的关键，后面会讲。
请求时求解：solve_dependencies()
请求进来后，FastAPI调用 solve_dependencies() 来实际执行依赖树：
async def solve_dependencies(
*,
request: Request,
dependant: Dependant,
body: Optional[Union[Dict[str, Any], FormData]] = None,
dependency_overrides_provider: Optional[Any] = None,
dependency_cache: Optional[Dict[Tuple[Callable[..., Any], Tuple[str]], Any]] = None,
) -> Tuple[Dict[str, Any], List[ErrorWrapper], Optional[Request]]:
values: Dict[str, Any] = {}
errors: List[ErrorWrapper] = []
if dependency_cache is None:
dependency_cache = {} # ★ 每个请求独立的缓存字典
# 1. 先递归求解所有子依赖
for sub_dependant in dependant.dependencies:
# 检查是否被 override
use_sub_dependant = sub_dependant
if dependency_overrides_provider and dependency_overrides_provider.dependency_overrides:
dependency_overrides = dependency_overrides_provider.dependency_overrides
if sub_dependant.call in dependency_overrides:
call = dependency_overrides[sub_dependant.call]
use_sub_dependant = get_dependant(path=sub_dependant.path, call=call)
solved_result = await solve_dependencies(
request=request,
dependant=use_sub_dependant,
body=body,
dependency_overrides_provider=dependency_overrides_provider,
dependency_cache=dependency_cache, # ★ 缓存向下传递
)
sub_values, sub_errors, _ = solved_result
# 2. 缓存命中则直接跳过执行
if (
use_sub_dependant.use_cache
and use_sub_dependant.cache_key
and use_sub_dependant.cache_key in dependency_cache
):
solved = dependency_cache[use_sub_dependant.cache_key]
else:
# 3. 执行依赖函数
solved = await call(**sub_values)
if sub_dependant.cache_key and use_sub_dependant.use_cache:
dependency_cache[sub_dependant.cache_key] = solved
values[sub_dependant.name] = solved
# 4. 求解当前节点自己的参数（path/query/header等）
# ...
return values, errors, request这段代码有三个关键点：
（1）dependency_cache 是请求级别的
每次请求进来，solve_dependencies 都会新建一个 dependency_cache = {}。这意味着：同一个请求内，相同依赖只执行一次；不同请求之间，完全隔离。这解决了请求间的状态污染问题——但我开头提到的线上事故，恰恰是因为某个依赖函数内部持有的是模块级或类级的可变状态，绕过了这个请求级缓存。
（2）cache_key 的构造
cache_key 在 get_dependant 中通过 get_dependant_cache_key 生成，逻辑大致是：把依赖函数的完整限定名 + 路径参数名列表拼成一个元组。这意味着：
- 同一个依赖函数，被两个不同的路由使用，缓存键不同（因为路径不同）
- 路径参数名不同，缓存键也不同
（3）缓存判断的顺序陷阱
注意代码里先判断 use_cache，再判断 cache_key in dependency_cache。如果有人把 Depends(get_db, use_cache=False) 写在两个位置，那么这两个位置会各执行一次 get_db。这在需要"每次都要新连接"的场景下是必要的，但也容易被误用。
yield 依赖的生命周期
FastAPI的生成器依赖（yield形式）是它的一大特色，用于管理资源（DB会话、文件句柄等）：
async def get_db():
db = SessionLocal()
try:
yield db # ← 请求处理函数在这里执行
finally:
db.close() # ← 请求处理完后执行清理这个yield的时机控制，藏在 fastapi/routing.py 的 run_endpoint_function 和依赖求解的 AsyncExitStack 中：
# fastapi/routing.py (简化)
async with AsyncExitStack() as async_exit_stack:
solved_result = await solve_dependencies(
request=request,
dependant=dependant,
body=body,
dependency_overrides_provider=self.dependency_overrides_provider,
async_exit_stack=async_exit_stack, # ★ 关键
)
values, errors, background_tasks, sub_response, _ = solved_result
# 执行路由处理函数
raw_response = await run_endpoint_function(
dependant=dependant, values=values, is_coroutine=is_coroutine
)
# 响应发送完毕后，AsyncExitStack 退出，触发所有 yield 依赖的 finallyAsyncExitStack 是Python标准库 contextlib 里的工具，它像一个栈式上下文管理器：所有yield依赖被压入栈，请求处理完成后按后进先出（LIFO）顺序退出。
这带来一个重要结论：yield依赖的清理代码（finally块）在响应返回给客户端之后才执行。如果你在finally里做了耗时操作（比如写日志到远程服务器），会拖慢响应。这是很多性能问题的隐形来源。
用餐厅类比：备菜员准备的食材，要等顾客吃完离店后才收拾，而不是菜一上桌就收拾。
实战代码
理论讲完，上三个可运行的例子，逐层加深。
示例1：带请求级缓存的依赖链
# demo1.py
from fastapi import FastAPI, Depends, Request
from typing import Annotated
import time
app = FastAPI()
# 模拟一个"昂贵"的配置加载
def get_settings():
print(f"[get_settings] 执行了！时间戳={time.time():.6f}")
time.sleep(0.1) # 模拟耗时
return {"db_url": "postgresql://localhost/app", "debug": False}
def get_db(settings: Annotated[dict, Depends(get_settings)]):
print(f"[get_db] 执行了！settings={settings['db_url']}")
return {"connection": "fake_conn", "settings": settings}
def get_current_user(
db: Annotated[dict, Depends(get_db)],
settings: Annotated[dict, Depends(get_settings)], # ← 与 get_db 依赖同一个 settings
):
print(f"[get_current_user] 执行了！")
return {"user_id": 42, "role": "admin"}
@app.get("/profile")
def read_profile(
user: Annotated[dict, Depends(get_current_user)],
db: Annotated[dict, Depends(get_db)], # ← 又一次依赖 get_db
):
return {"user": user, "db_conn": db["connection"]}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, port=8000)启动后访问 http://localhost:8000/profile，控制台输出：
[get_settings] 执行了！时间戳=...
[get_db] 执行了！settings=postgresql://localhost/app
[get_current_user] 执行了！注意：get_settings 只执行了一次，get_db 也只执行了一次！ 尽管它们被多个地方依赖。这就是请求级缓存的威力。如果把 Depends(get_settings) 改成 Depends(get_settings, use_cache=False)，你会看到它被执行多次。
验证一下：在同一请求中，read_profile 里的 db 和 get_current_user 里的 db 是同一个对象（内存地址相同）。这个特性在传递DB会话时至关重要。
示例2：yield依赖的资源管理
# demo2.py
from fastapi import FastAPI, Depends, HTTPException
from typing import Annotated, Generator
import time
app = FastAPI()
class FakeSession:
def __init__(self):
self.closed = False
print(f"[FakeSession] 创建，id={id(self)}")
def query(self, sql: str):
if self.closed:
raise RuntimeError("Session already closed!")
return f"result of: {sql}"
def close(self):
self.closed = True
print(f"[FakeSession] 关闭，id={id(self)}")
def get_session() -> Generator[FakeSession, None, None]:
session = FakeSession()
try:
yield session # 请求处理阶段
finally:
# 无论请求成功还是抛异常，都会执行
session.close()
@app.get("/users/{user_id}")
def get_user(user_id: int, session: Annotated[FakeSession, Depends(get_session)]):
data = session.query(f"SELECT * FROM users WHERE id={user_id}")
return {"data": data, "session_id": id(session)}
@app.get("/boom")
def boom(session: Annotated[FakeSession, Depends(get_session)]):
# 故意抛异常，验证 finally 是否执行
raise HTTPException(status_code=500, detail="boom")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, port=8000)访问 /users/1，输出：
[FakeSession] 创建，id=140234...
[FakeSession] 关闭，id=140234...访问 /boom：
[FakeSession] 创建，id=140234...
[FakeSession] 关闭，id=140234...即使路由函数抛异常，finally 依然执行。这是AsyncExitStack的保证。很多开发者担心"异常时DB连接会不会泄漏"，答案是：只要你在yield依赖里用try/finally，就不会。
但是：如果 yield 之后的清理代码本身抛异常，会覆盖掉原来的响应。这是个大坑，后面会讲。
示例3：多租户场景下的上下文隔离（踩坑示范）
这个例子复现我开头提到的线上事故，并给出正确解法。
# demo3.py — 反例：模块级可变状态
from fastapi import FastAPI, Depends, Header
from typing import Annotated
import asyncio
app = FastAPI()
# ❌ 错误示范：模块级可变状态
_current_tenant = {"id": None}
def get_tenant(tenant_id: Annotated[str, Header(alias="X-Tenant-Id")]):
# 高并发下，多个协程会互相覆盖这个全局字典
_current_tenant["id"] = tenant_id
return _current_tenant
@app.get("/bad/data")
async def bad_data(tenant: Annotated[dict, Depends(get_tenant)]):
await asyncio.sleep(0.05) # 模拟IO等待，期间其他请求会修改 _current_tenant
return {"tenant_seen": tenant["id"]}测试一下：并发发送两个请求，Header 分别是 X-Tenant-Id: A 和 X-Tenant-Id: B。你会看到响应里的 tenant_seen 可能都是 B——串号了。
正确解法有三种：
# ✅ 解法1：返回不可变的、请求级的数据副本
def get_tenant_v1(tenant_id: Annotated[str, Header(alias="X-Tenant-Id")]):
return {"id": tenant_id} # 每次返回新字典，天然隔离
# ✅ 解法2：用 contextvars 保证协程级隔离
import contextvars
_tenant_ctx: contextvars.ContextVar[str] = contextvars.ContextVar("tenant")
def get_tenant_v2(tenant_id: Annotated[str, Header(alias="X-Tenant-Id")]):
token = _tenant_ctx.set(tenant_id)
# 注意：这里没地方 reset，需要 yield 依赖
return tenant_id
# ✅ 解法3（推荐）：yield 依赖 + contextvars，保证清理
def get_tenant_v3(tenant_id: Annotated[str, Header(alias="X-Tenant-Id")]):
token = _tenant_ctx.set(tenant_id)
try:
yield tenant_id
finally:
_tenant_ctx.reset(token) # 请求结束，还原上下文
@app.get("/good/data")
async def good_data(tenant: Annotated[str, Depends(get_tenant_v3)]):
await asyncio.sleep(0.05)
return {"tenant_seen": _tenant_ctx.get()}核心原则：依赖函数要么返回不可变数据，要么用contextvars做协程隔离，要么用yield管理生命周期。永远不要在依赖里写模块级的可变状态。
方案对比：FastAPI DI vs 其他框架
| 维度 | FastAPI | Spring (Java) | Django | Flask |
|---|---|---|---|---|
| DI核心机制 | 函数签名内省 + 运行时解析 | 注解 + 反射 + 启动时构建容器 | 无内置DI（靠中间件/装饰器） | 无内置DI（靠扩展） |
| 容器构建时机 | 应用启动时构建依赖树 | 应用启动时构建Bean容器 | N/A | N/A |
| 依赖解析时机 | 请求时按需执行 | 单例启动即创建 / Prototype按需 | N/A | N/A |
| 作用域 | 请求级（默认）+ 可关闭缓存 | Singleton/Prototype/Request/Session | N/A | N/A |
| 缓存粒度 | 请求内 + 每个依赖函数 | 全局单例或每次新建 | N/A | N/A |
| 异步支持 | 原生async/await | 需Reactor/WebFlux | 3.1+支持async视图 | 弱 |
| 学习曲线 | 低（声明即用） | 高（注解体系庞大） | 中 | 低 |
| 循环依赖检测 | 无显式检测（会栈溢出） | 启动时报错 | N/A | N/A |
FastAPI DI的独特之处：
- 轻量到极致：没有Bean容器，没有XML配置，没有代理对象。依赖就是普通函数。
- 请求级缓存是默认行为：Spring默认单例，容易造成状态污染；FastAPI默认请求级，更安全。
- 没有循环依赖检测：这是FastAPI的一个短板。如果A依赖B、B依赖A，你会在启动时收到
RecursionError而不是友好的错误信息。Spring会明确告诉你循环依赖的路径。 - 依赖覆盖（dependency_overrides）：这是FastAPI测试友好性的关键，可以轻松mock掉任何依赖。
# 测试时覆盖依赖
app.dependency_overrides[get_db] = lambda: {"connection": "mock_conn"}相比Django的中间件方案，FastAPI的DI更显式、更可组合；相比Spring，它牺牲了一些企业级特性（AOP、事务传播等）换取了开发效率。
最佳实践与避坑指南
坑1：yield依赖的清理异常会污染响应
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close() # 如果这里抛异常，客户端会收到500，即使路由函数成功了解法：清理逻辑用 try/except 包住，异常只记日志不抛出。
def get_db():
db = SessionLocal()
try:
yield db
finally:
try:
db.close()
except Exception:
logger.exception("关闭DB连接失败")坑2：误用 use_cache=False 导致资源爆炸
# ❌ 每个依赖它的地方都会新建一个连接
def get_db():
...
return SessionLocal()
@app.get("/a")
def a(db1 = Depends(get_db, use_cache=False), db2 = Depends(get_db, use_cache=False)):
... # db1 和 db2 是两个不同的连接！实践：除非你明确需要"每次调用都是新实例"（比如生成随机数、记录调用次数），否则永远用默认的use_cache=True。
坑3：依赖函数签名里的参数被误当成Query参数
def get_user(user_id: str): # FastAPI会把 user_id 当成 query 参数！
...FastAPI会尝试从请求的query string里解析user_id。如果你的本意是"这是个内部参数"，必须显式声明：
def get_user(user_id: str = Depends(get_user_id_from_token)):
...坑4：循环依赖导致栈溢出
def a(b = Depends(lambda: None)): ...
# A依赖B，B依赖A → 启动时 RecursionError实践：依赖关系保持单向。如果确实需要互相引用，提取公共部分到第三个依赖，或者用Request对象直接传递状态。
坑5：在依赖里做重量级初始化
# ❌ 每次请求都重新加载ML模型
def get_model():
return load_heavy_model() # 耗时2秒解法：用应用启动事件加载到app.state，依赖里只取引用：
@app.on_event("startup")
async def startup():
app.state.model = load_heavy_model()
def get_model(request: Request):
return request.app.state.model最佳实践清单
- 依赖函数保持纯粹：输入输出明确，不持有可变状态
- 资源管理用
yield：DB会话、文件、锁，都用生成器依赖 - 测试用
dependency_overrides：不要在业务代码里写if-else判断环境 - 依赖分层：底层（配置/连接）→ 中层（业务服务）→ 顶层（路由）
- 善用
Annotated：UserId = Annotated[str, Depends(get_user_id)]，让签名更清晰 - 警惕
contextvars：跨协程的状态必须用它，且配合yield做reset
总结
FastAPI的依赖注入系统用"函数签名内省 + 请求级缓存 + AsyncExitStack"三件套，实现了一个轻量但强大的DI框架。回顾本文要点：
Depends只是声明载体，真正的解析在get_dependant（启动时构建树）和solve_dependencies（请求时执行）中完成
- 请求级缓存是默认行为，同一个请求内相同依赖只执行一次，
dependency_cache字典在每次请求时新建
yield依赖的生命周期由AsyncExitStack管理，清理在响应发送后按LIFO顺序执行
- 模块级可变状态是万恶之源，用
contextvars或返回不可变数据做隔离
- FastAPI没有循环依赖检测，设计依赖树时要保持单向
延伸思考：FastAPI的DI设计哲学是"约定优于配置、显式优于隐式"，这与Spring的"配置驱动"路线截然不同。在微服务、Serverless等场景下，这种轻量设计更有优势；但在需要复杂事务管理、AOP切面的企业级应用中，你可能需要引入额外的库（如dependency-injector）来补充。
最后送一句话给正在用FastAPI的同行：依赖注入不是为了少写代码，而是为了让状态边界清晰可见。当你理解了每个依赖的生命周期，你就掌握了FastAPI的半壁江山。