依赖注入(Dependency Injection, DI)是 FastAPI 解耦架构的核心。本文将深入 DI 的执行原理,重点分析带有
yield 的依赖项如何优雅地管理资源(如数据库连接)的生命周期,以及如何在类与函数之间选择合适的依赖形式。1. 依赖注入的设计哲学#
在传统开发中,常在函数内部手动初始化资源(如 db = Session()),这导致代码耦合度高、难以测试。FastAPI 的 DI 系统允许将“需要什么”声明在函数参数中,由框架负责“如何构建”和“何时注入”。
1.1 Depends 的工作流#
当请求到达路由时,FastAPI 执行以下步骤:
- 解析路由函数签名的
Depends参数。 - 查找该依赖项是否已在缓存中(默认
use_cache=True)。 - 如果未缓存,执行依赖函数,获取返回值。
- 将返回值注入到路由函数中。
- 关键点:如果依赖项依赖于其他依赖项,FastAPI 会构建“依赖图(Dependency Graph)”,并按拓扑顺序自底向上执行。
2. 资源生命周期管理:Yield 依赖#
这是工程实践中最重要的模式。对于数据库会话、网络连接等需要“创建 -> 使用 -> 销毁”的资源,普通的 return 依赖无法处理“销毁”逻辑。FastAPI 借鉴了 pytest 的 fixture 机制,支持包含 yield 的依赖项。
2.1 数据库 Session 管理的标准写法#
# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine("postgresql://user:pass@localhost/db")
SessionLocal = sessionmaker(bind=engine)
# 依赖项定义
def get_db():
db = SessionLocal() # 1. Setup: 创建连接
try:
yield db # 2. Inject: 注入给路由函数,并暂停执行
finally:
db.close() # 3. Teardown: 路由执行完毕后(无论成功或异常),恢复执行关闭连接
2.2 在路由中使用#
from sqlalchemy.orm import Session
from fastapi import Depends
@app.get("/users/")
async def read_users(db: Session = Depends(get_db)):
# 此时 db 是一个打开的数据库会话
users = db.query(User).all()
return users
# 函数返回后,get_db 中的 finally 块被触发,连接被归还/关闭
这种模式完美替代了传统的中间件或装饰器,且具有类型感知能力,是 FastAPI 处理事务范围(Transaction Scope)的标准方式。
3. 依赖形式:函数 vs 类#
FastAPI 允许使用函数或类作为依赖项。两者在底层机制上一致(都是 Callable),但在工程组织上各有优势。
3.1 函数依赖 (Function Dependency)#
适用于简单的逻辑复用,如获取当前用户、提取分页参数。
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
return commons
3.2 类依赖 (Class Dependency)#
当依赖项参数较多,或需要利用继承特性时,使用类更为清晰。
class CommonQueryParams:
def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]):
# 此处 commons 是 CommonQueryParams 的实例
# 享有完整的 IDE 属性提示
return commons.dict()
- 简写技巧:如果依赖项是类,可以简写为
commons: Annotated[CommonQueryParams, Depends()],FastAPI 会自动推断使用CommonQueryParams类本身。
4. 全局依赖与子依赖#
4.1 全局依赖 (Global Dependencies)#
某些逻辑(如 API Key 校验、IP 白名单)需要应用于整个应用。可以在 FastAPI 或 APIRouter 初始化时注入。
async def verify_token(x_token: str = Header()):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
# 应用于整个 App
app = FastAPI(dependencies=[Depends(verify_token)])
# 或应用于特定 Router
user_router = APIRouter(dependencies=[Depends(verify_token)])
注意:全局依赖的返回值通常会被忽略,它们主要用于执行副作用(Side Effects)或抛出异常。
4.2 缓存机制 (Cache)#
默认情况下,如果一个请求的依赖图中多次出现同一个依赖项,FastAPI 只会执行一次,并在同一次请求中复用返回值。
- 场景:
get_current_user依赖get_token,get_user_permissions也依赖get_token。FastAPI 保证get_token只执行一次。 - 禁用缓存:如果依赖项需要每次都重新计算(不常见),可设置
Depends(dep, use_cache=False)。
总结#
- 资源管理:必须掌握
yield依赖模式,它是数据库连接、Redis 客户端等资源安全关闭的基石。 - 依赖图:理解依赖的传递性和拓扑执行顺序,有助于设计复杂的权限验证系统。
- 类型选择:简单逻辑用函数,复杂参数组合用类,保持代码结构的清晰。
- 复用策略:利用
use_cache=True(默认)避免重复计算,提高单次请求的处理效率。