FastAPI 教程

Chapter 03 · 依赖注入:代码的乐高积木

📚 学完本章,你将能够:

  • 解释"依赖注入"解决了什么实际问题(告别复制粘贴)
  • 用 Depends 把公共逻辑抽象成可复用函数
  • 理解子依赖链:依赖的依赖的依赖
  • 用 yield 管理资源的创建和释放(如数据库连接)
  • 在测试中替换依赖,实现不启动数据库的单元测试

💡 场景引入

你的图书 API 有 10 个接口需要"当前登录用户"信息。你不得不在每个接口里重复写一遍"从请求头读取 Token → 解析 Token → 查数据库确认用户存在"的流程。

更麻烦的是,如果 Token 的解析逻辑变了(比如换了加密算法),你得改 10 个地方。依赖注入就是来解决这种重复问题的。

3.1 没有依赖注入的噩梦

❌ 到处复制粘贴
@app.get("/books")
async def list_books(authorization: str = Header()):
    token = authorization.replace("Bearer ", "")
    user = verify_token(token)  # 同样的代码
    if not user:
        raise HTTPException(401)
    return books

@app.post("/books")
async def create_book(data: BookCreate, authorization: str = Header()):
    token = authorization.replace("Bearer ", "")
    user = verify_token(token)  # 又写了一遍!
    if not user:
        raise HTTPException(401)
    ...

@app.delete("/books/{book_id}")
async def delete_book(book_id: int, authorization: str = Header()):
    token = authorization.replace("Bearer ", "")
    user = verify_token(token)  # 第三遍!!
    if not user:
        raise HTTPException(401)
    ...

这段代码有三个问题:

  1. 重复:同样的逻辑写了 N 遍
  2. 难维护:改一处逻辑要改 N 个地方
  3. 难测试:每个接口都耦合了 Token 解析,测试时必须走完整流程

3.2 Depends:把公共逻辑抽出来

FastAPI 的 Depends 让你把公共逻辑写成一个函数,然后在需要的地方"注入":

✅ 使用依赖注入
from fastapi import Depends, HTTPException, Header
from typing import Annotated

async def get_current_user(authorization: Annotated[str, Header()]) -> User:
    token = authorization.replace("Bearer ", "")
    user = verify_token(token)
    if not user:
        raise HTTPException(401, detail="Invalid token")
    return user

@app.get("/books")
async def list_books(user: Annotated[User, Depends(get_current_user)]):
    # user 已经是验证通过的用户对象
    return books

@app.post("/books")
async def create_book(data: BookCreate, user: Annotated[User, Depends(get_current_user)]):
    ...

@app.delete("/books/{book_id}")
async def delete_book(book_id: int, user: Annotated[User, Depends(get_current_user)]):
    ...

现在 Token 解析逻辑只写在一处。所有接口都通过 Depends(get_current_user) 复用这段逻辑。如果以后要改解析方式,只改 get_current_user 一个函数即可。

依赖注入的工作方式
浏览器请求
GET /books
执行依赖
get_current_user
进入视图函数
list_books(user)

3.3 子依赖链

依赖函数本身也可以依赖其他依赖函数。这叫做子依赖。

三级依赖链
# 第一层:从 Header 提取 Token
async def get_token(auth: Annotated[str, Header(alias="Authorization")]) -> str:
    if not auth.startswith("Bearer "):
        raise HTTPException(401, "Must use Bearer token")
    return auth[7:]

# 第二层:验证 Token 得到用户
async def get_current_user(token: Annotated[str, Depends(get_token)]) -> User:
    user = verify_jwt(token)
    if not user:
        raise HTTPException(401, "Invalid or expired token")
    return user

# 第三层:检查是否是管理员
async def require_admin(user: Annotated[User, Depends(get_current_user)]) -> User:
    if user.role != "admin":
        raise HTTPException(403, "Admin only")
    return user

# 使用
@app.delete("/users/{user_id}")
async def ban_user(user: Annotated[User, Depends(require_admin)]):
    ...

当请求进入 /users/5 时,FastAPI 会自动按顺序执行:

  1. get_token → 提取字符串
  2. get_current_user → 验证并得到 User 对象
  3. require_admin → 检查角色
  4. 最后进入 ban_user 视图函数
依赖缓存

同一个请求中,如果多个参数都 Depends(get_current_user),FastAPI 只会执行一次 get_current_user,然后缓存结果复用。这避免了重复查数据库。

3.4 yield:管理资源的创建和释放

数据库连接、文件句柄这类资源需要"用完后关闭"。yield 让依赖函数变成上下文管理器:

yield 依赖
async def get_db():
    db = create_database_connection()  # 请求开始时创建连接
    try:
        yield db  # 把连接交给视图函数使用
    finally:
        db.close()  # 请求结束后关闭连接,即使出异常也会执行

yield 之前的代码在视图函数前执行,yield 之后的代码在视图函数后执行(包括异常时)。这非常适合数据库会话管理:

数据库会话依赖(预告:下一章详解)
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

async_session = async_sessionmaker(engine, expire_on_commit=False)

async def get_db():
    async with async_session() as session:
        yield session  # 视图函数使用 session
        # 退出 async with 时自动关闭连接

3.5 依赖覆盖:测试的杀手锏

这是依赖注入最强大的特性。测试中你可以临时替换任何一个依赖,不用修改业务代码。

测试覆盖
from fastapi.testclient import TestClient

# 测试中不用真的去验证 JWT
async def mock_current_user():
    return User(id=1, username="tester", role="admin")

# 把真实的 get_current_user 换成假的
app.dependency_overrides[get_current_user] = mock_current_user

client = TestClient(app)

def test_delete_book():
    response = client.delete("/books/1")
    assert response.status_code == 204

# 测试结束后清理
app.dependency_overrides.clear()

这意味着:测试认证接口时,你不需要真的生成 Token;测试数据库操作时,你可以换成内存数据库——这些都会在第 9 章详细展开。

🎯 知识检测

问题 1
Depends 的核心作用是什么?
让代码运行得更快
把公共逻辑抽成可复用函数,自动注入到路由中
自动生成数据库表
替代 try/except 处理异常
解析:Depends 解决的是"代码重复"问题。你把公共逻辑(如获取当前用户、打开数据库连接)写成函数,FastAPI 会在请求到达路由前自动调用它,并把结果注入参数。
问题 2
以下关于 yield 依赖的说法,哪个是正确的?
yield 之前的代码在响应发送后执行
yield 之后的代码在视图函数结束后执行,即使发生异常
yield 依赖不能返回数据给视图函数
yield 只能在同步函数中使用
解析:yield 把依赖分成两段:前段在路由前执行(如创建连接),后段在路由后执行(如关闭连接)。后段在 finally 语义下执行,保证资源一定被释放。它可以返回数据(yield 后面的值就是注入的值),且支持 async。
问题 3
在测试中,app.dependency_overrides[get_db] = fake_db 的作用是什么?
永久修改了应用的数据库配置
临时替换 get_db 依赖,测试结束后可以清除
只对 GET 请求生效
会自动回滚数据库事务
解析:dependency_overrides 是一个字典,只在当前应用实例中临时替换依赖。测试结束后调用 .clear() 即可恢复,不会污染其他测试或生产代码。

本章小结

✓

Depends 消灭重复代码

把"获取当前用户""打开数据库"这类公共逻辑写成函数,到处复用。

✓

依赖可以无限嵌套

get_token → get_current_user → require_admin,像搭积木一样组合权限。

✓

yield 管理资源生命周期

连接数据库 → yield 给路由使用 → 路由结束关闭连接。保证不泄漏。

✓

依赖覆盖让测试更轻松

测试中替换真实依赖为 Mock,不用真的连数据库或发邮件。

🛠️ 动手实验

实现一个三层依赖链:get_token(从 Header 提取)→ get_current_user(验证用户存在)→ require_active_user(检查用户未被冻结)。
创建一个只有登录用户能访问的接口和一个只有管理员能访问的接口。测试时使用 dependency_overrides 替换认证依赖。