FastAPI: 依赖注入系统 — FastAPI 的灵魂设计
最后更新：2026-08-26
依赖注入就像乐高积木的接口——每块积木只关心自己该提供什么，不需要知道谁在用它。DB 连接、当前用户、权限检查，层层嵌套自动解析。
1. 你将学到
Depends()基础：函数依赖、类依赖、子依赖嵌套- 依赖的生命周期：请求级 vs 应用级（
yield依赖与资源清理） - 全局依赖与路由组依赖：
dependencies=[Depends(...)] - 依赖覆盖与测试：
app.dependency_overrides实战技巧 - Alice 场景：PriceTracker 的 DB Session 依赖、当前用户依赖、权限检查依赖——三层嵌套 DI 链
2. Alice 的真实故事
(1) 痛点：每个端点重复获取 DB 和用户
Alice 的 PriceTracker 有 20 个端点，每个都需要打开数据库连接、验证 JWT Token、获取当前用户、检查权限。她在每个端点函数里复制粘贴同样的 15 行初始化代码，一旦 DB 连接方式变化，需要修改 20 个地方。
(2) 依赖注入的解法
FastAPI 的依赖注入将重复逻辑提取为可复用的依赖函数，端点只需声明 db = Depends(get_db)，FastAPI 自动解析和注入，包括子依赖的递归解析。
PYTHON
from fastapi import Depends
async def get_db():
db = Database()
yield db
db.close()
async def get_current_user(token: str, db=Depends(get_db)):
return verify_token(token, db)
@app.get("/products")
async def list_products(user=Depends(get_current_user), db=Depends(get_db)):
# user and db are auto-injected
...
(3) 收益
DB 连接、用户验证、权限检查从 20 个端点各写 15 行变成 3 个依赖函数，修改一处全局生效。测试时只需 app.dependency_overrides[get_db] = lambda: mock_db，一行替换所有端点的数据库。
3. Depends() 基础
(1) 函数依赖
▶ 示例：简单函数依赖
PYTHON
from fastapi import FastAPI, Depends
app = FastAPI()
def common_parameters(
q: str | None = None,
skip: int = 0,
limit: int = 100,
):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/products")
async def list_products(commons: dict = Depends(common_parameters)):
return commons
@app.get("/prices")
async def list_prices(commons: dict = Depends(common_parameters)):
return commons
输出：
TEXT
📖 仅展示
# 函数定义成功
▶ 示例：类依赖
PYTHON
from fastapi import FastAPI, Depends, Query
class CommonQueryParams:
def __init__(
self,
q: str | None = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=200),
):
self.q = q
self.skip = skip
self.limit = limit
app = FastAPI()
@app.get("/products")
async def list_products(commons: CommonQueryParams = Depends(CommonQueryParams)):
return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}
输出：
TEXT
📖 仅展示
# 函数定义成功
(2) 函数依赖 vs 类依赖
| 维度 | 函数依赖 | 类依赖 |
|---|---|---|
| 定义方式 | def get_xxx() |
class Xxx: + __init__ |
| 适用场景 | 简单逻辑、单次调用 | 需要状态、多方法 |
| 参数解析 | 函数参数自动解析 | __init__ 参数自动解析 |
| 复用性 | 高 | 中 |
| 推荐度 | 首选 | 复杂场景使用 |
4. 子依赖嵌套
(1) 依赖解析树
graph TD
A[get_current_user] --> B[get_db]
A --> C[get_token_from_header]
C --> D[OAuth2PasswordBearer]
E[require_admin] --> A
E --> F[check_subscription]
style A fill:#e1f5fe
style E fill:#fff3e0
FastAPI 递归解析依赖树：require_admin → get_current_user → get_db，每个依赖只实例化一次（同一请求内缓存）。
▶ 示例：子依赖嵌套——DB → User
PYTHON
from fastapi import FastAPI, Depends, HTTPException, Header
app = FastAPI()
# Level 1: Database session
async def get_db():
db = {"connection": "active"}
try:
yield db
finally:
db["connection"] = "closed"
# Level 2: Get current user (depends on DB)
async def get_current_user(
authorization: str = Header(...),
db: dict = Depends(get_db),
):
if not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Invalid token")
token = authorization.replace("Bearer ", "")
# Simplified: in production, verify JWT
user = {"id": 1, "username": "alice", "role": "admin"}
return user
# Level 3: Require admin (depends on current user)
async def require_admin(user: dict = Depends(get_current_user)):
if user["role"] != "admin":
raise HTTPException(status_code=403, detail="Admin required")
return user
@app.get("/admin/stats")
async def admin_stats(admin: dict = Depends(require_admin)):
return {"admin": admin["username"], "total_products": 1000000}
输出：
TEXT
📖 仅展示
# 函数定义成功
5. 依赖生命周期与 yield
(1) 请求级 vs 应用级
| 生命周期 | 声明方式 | 作用域 | 典型用途 |
|---|---|---|---|
| 请求级 | def dep(): yield x; cleanup |
每次请求创建+销毁 | DB Session、Redis 连接 |
| 应用级 | @app.on_event("startup") |
应用启动时创建 | 引擎初始化、连接池 |
▶ 示例：yield 依赖与资源清理
PYTHON
from fastapi import FastAPI, Depends
app = FastAPI()
# Request-scoped dependency with cleanup
async def get_db_session():
# Setup: create session
session = {"id": "session-123", "active": True}
print(f"DB session opened: {session['id']}")
try:
yield session # This is injected into the endpoint
finally:
# Cleanup: close session (runs after response)
session["active"] = False
print(f"DB session closed: {session['id']}")
@app.get("/products")
async def list_products(db: dict = Depends(get_db_session)):
print(f"Using DB session: {db['id']}")
return {"session_id": db["id"]}
输出（服务端日志）：
TEXT
📖 仅展示
DB session opened: session-123
Using DB session: session-123
DB session closed: session-123
6. 全局依赖与路由组依赖
(1) 不同范围的依赖声明
▶ 示例：路由组依赖
PYTHON
from fastapi import FastAPI, Depends, APIRouter, Header, HTTPException
async def verify_api_key(x_api_key: str = Header(...)):
if x_api_key != "secret-key-123":
raise HTTPException(status_code=401, detail="Invalid API key")
return x_api_key
app = FastAPI()
# Public routes - no auth required
public_router = APIRouter()
@public_router.get("/health")
async def health():
return {"status": "healthy"}
# Protected routes - API key required for all routes in this group
protected_router = APIRouter(dependencies=[Depends(verify_api_key)])
@protected_router.get("/products")
async def list_products():
return [{"id": 1, "name": "Widget"}]
@protected_router.post("/products")
async def create_product():
return {"id": 2, "name": "New Product"}
# Register routers
app.include_router(public_router)
app.include_router(protected_router, prefix="/api/v1")
输出：
TEXT
📖 仅展示
# 函数定义成功
| 依赖范围 | 声明位置 | 影响范围 |
|---|---|---|
| 端点级 | @app.get("/", dependencies=[...]) |
单个端点 |
| 路由组级 | APIRouter(dependencies=[...]) |
路由组所有端点 |
| 全局级 | FastAPI(dependencies=[...]) |
应用所有端点 |
7. 依赖覆盖与测试
(1) dependency_overrides
测试时需要替换真实依赖（如数据库连接、外部 API），app.dependency_overrides 允许在不修改源码的情况下替换依赖实现。
▶ 示例：测试中替换数据库依赖
PYTHON
from fastapi.testclient import TestClient
from fastapi import FastAPI, Depends
app = FastAPI()
# Real dependency
async def get_db():
return {"type": "postgresql", "host": "prod-db"}
# Mock dependency for testing
def get_mock_db():
return {"type": "sqlite", "host": "memory"}
@app.get("/db-info")
async def db_info(db: dict = Depends(get_db)):
return db
# In tests:
def test_db_info():
app.dependency_overrides[get_db] = get_mock_db
client = TestClient(app)
response = client.get("/db-info")
assert response.json() == {"type": "sqlite", "host": "memory"}
# Clean up
app.dependency_overrides.clear()
输出：
TEXT
📖 仅展示
# 函数定义成功
❓ 常见问题
Q Depends 的依赖会被多次调用吗？
A 同一请求内，同一个依赖只执行一次，结果被缓存复用。不同请求会重新执行。
Q yield 依赖的 cleanup 什么时候执行？
A 在响应发送给客户端之后执行。如果端点抛出异常，cleanup 仍然执行（类似 try/finally）。
Q 依赖能嵌套多深？
A 没有硬性限制，但超过 3 层嵌套会增加调试难度。PriceTracker 的三层 DI 链（get_db → get_current_user → require_admin）是推荐的上限。
Q 同步和异步依赖能混用吗？
A 可以。FastAPI 自动处理同步/异步依赖。同步依赖在线程池中执行，异步依赖在事件循环中执行。
Q dependency_overrides 会影响其他测试吗？
A 会，因为是修改 app 对象。务必在测试结束后调用
app.dependency_overrides.clear() 或使用 fixture 管理。Q 类依赖的 init 参数怎么注入？
A 和函数依赖一样，init
📖 小节
Depends()将重复逻辑提取为可复用的依赖函数或类，端点只需声明不需要关心实现- 子依赖自动递归解析，同请求内缓存，避免重复执行
yield依赖实现请求级生命周期管理，确保资源创建和清理配对- 路由组依赖 (
APIRouter(dependencies=[...])) 让一组端点共享认证/校验逻辑 app.dependency_overrides是测试利器，一行代码替换所有端点的真实依赖
📝 作业
- 基础题（难度⭐）：创建
get_db依赖函数，返回一个字典模拟数据库连接，在两个端点中注入并使用。提示：db: dict = Depends(get_db) - 进阶题（难度⭐⭐）：实现两层嵌套依赖：
get_db→get_current_user（从请求头读 Authorization），在/me端点注入当前用户信息。提示：user: dict = Depends(get_current_user) - 挑战题（难度⭐⭐⭐）：实现 PriceTracker 三层 DI 链：
get_db（yield + cleanup）→get_current_user（验证 token）→require_subscription("pro")（检查订阅级别），并用dependency_overrides测试时替换 DB 依赖。提示：yield依赖 +app.dependency_overrides[get_db] = mock_fn
---|
