跳过正文
FastAPI:依赖注入系统
  1. Blog/

FastAPI:依赖注入系统

·1463 字·3 分钟
目录
FastAPI - 这篇文章属于一个选集。
§ 5: 本文
依赖注入(Dependency Injection, DI)是 FastAPI 解耦架构的核心。本文将深入 DI 的执行原理,重点分析带有 yield 的依赖项如何优雅地管理资源(如数据库连接)的生命周期,以及如何在类与函数之间选择合适的依赖形式。

1. 依赖注入的设计哲学
#

在传统开发中,常在函数内部手动初始化资源(如 db = Session()),这导致代码耦合度高、难以测试。FastAPI 的 DI 系统允许将“需要什么”声明在函数参数中,由框架负责“如何构建”和“何时注入”。

1.1 Depends 的工作流
#

当请求到达路由时,FastAPI 执行以下步骤:

  1. 解析路由函数签名的 Depends 参数。
  2. 查找该依赖项是否已在缓存中(默认 use_cache=True)。
  3. 如果未缓存,执行依赖函数,获取返回值。
  4. 将返回值注入到路由函数中。
  5. 关键点:如果依赖项依赖于其他依赖项,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 白名单)需要应用于整个应用。可以在 FastAPIAPIRouter 初始化时注入。

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_tokenget_user_permissions 也依赖 get_token。FastAPI 保证 get_token 只执行一次。
  • 禁用缓存:如果依赖项需要每次都重新计算(不常见),可设置 Depends(dep, use_cache=False)

总结
#

  1. 资源管理:必须掌握 yield 依赖模式,它是数据库连接、Redis 客户端等资源安全关闭的基石。
  2. 依赖图:理解依赖的传递性和拓扑执行顺序,有助于设计复杂的权限验证系统。
  3. 类型选择:简单逻辑用函数,复杂参数组合用类,保持代码结构的清晰。
  4. 复用策略:利用 use_cache=True(默认)避免重复计算,提高单次请求的处理效率。
FastAPI - 这篇文章属于一个选集。
§ 5: 本文