Skip to content

第 10 章 · 依赖注入系统基础

本章目标:理解依赖注入解决什么问题,学会用 Depends 声明依赖、用 Annotated 类型别名复用依赖,并掌握同步/异步依赖混用与请求级缓存行为。

10.1 依赖注入解决什么问题

"依赖注入"(Dependency Injection)指:路径操作函数只声明自己需要什么,由框架负责构造并注入。官方列出的典型用途:

  • 多个接口共享同一段逻辑(分页参数解析、公共查询处理);
  • 共享数据库连接/会话;
  • 强制执行安全、认证、角色检查。

没有 DI 时的写法是每个接口复制粘贴同一段代码;有了 DI,逻辑写一次,FastAPI 在每次请求时自动调用并把结果塞进参数。

10.2 第一个依赖:Depends

依赖("dependable")就是一个普通函数,它的参数与路径操作函数完全同构:

python
from typing import Annotated
from fastapi import FastAPI, Depends

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    # 依赖函数:声明 q/skip/limit 三个查询参数并打包返回
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def list_items(
    commons: Annotated[dict, Depends(common_parameters)],  # 声明依赖
):
    return {"items": ["a", "b"], **commons}


@app.get("/users/")
async def list_users(
    commons: Annotated[dict, Depends(common_parameters)],  # 同一依赖复用
):
    return {"users": ["u1"], **commons}

关键规则:传给 Depends() 的是函数本身,不要加括号调用(写 Depends(common_parameters) 而不是 Depends(common_parameters()))。

每次请求到达时,FastAPI 会:

  1. 用正确的参数调用 common_parameters()(其中 q/skip/limit 被当作查询参数解析和校验);
  2. 把返回值赋给路径函数的 commons 参数。

注意:依赖里声明的 q/skip/limit 会自动进入 OpenAPI 文档,/docs 里能看到并调试它们。

10.3 用 Annotated 类型别名消除重复

上面两个接口都写了一遍 Annotated[dict, Depends(common_parameters)],还是有重复。由于 Annotated 是纯 Python 标准语法,可以把它存成类型别名变量

python
from typing import Annotated
from fastapi import FastAPI, Depends

Commons = Annotated[dict, Depends(common_parameters)]  # 类型别名,非 FastAPI 特性

app = FastAPI()


@app.get("/items/")
async def list_items(commons: Commons):     # 一行搞定
    return {"items": ["a"], **commons}


@app.get("/users/")
async def list_users(commons: Commons):
    return {"users": ["u1"], **commons}

官方从 0.95.0 起推荐这种写法。相比老式默认值写法(commons: dict = Depends(common_parameters)),Annotated 的优势是:类型信息完整保留,编辑器补全、mypy 静态检查都正常工作,且在大型代码库中复用别名极其方便。

10.4 同步与异步依赖可以自由混用

依赖和路径函数一样,async def 与普通 def 均可,且可以任意组合——async def 路径函数可以用同步依赖,反之亦然,FastAPI 会自动选择正确的事件循环/线程池执行方式。

选择原则与第 16 章一致:

  • 依赖内部是 I/O 且有异步客户端(如 AsyncSession)→ async def
  • 依赖是纯计算或调用同步阻塞库(如老版 psycopg2)→ 普通 def(避免阻塞事件循环)。
python
import time
from fastapi import Depends, FastAPI

app = FastAPI()


def sync_dep() -> dict:
    time.sleep(0.01)          # 模拟同步阻塞调用,FastAPI 会放到线程池执行
    return {"source": "sync"}


async def async_dep() -> dict:
    # 真实场景:await asyncpg 连接查询等
    return {"source": "async"}


@app.get("/mixed/")
async def mixed(
    a: Annotated[dict, Depends(sync_dep)],
    b: Annotated[dict, Depends(async_dep)],
):
    return {"a": a, "b": b}

10.5 请求级缓存:use_cache

同一个请求内,如果多个依赖(或子依赖)声明了同一个依赖函数,FastAPI 默认只调用它一次,结果缓存在该请求的上下文中复用:

python
from typing import Annotated
from fastapi import Depends, FastAPI

app = FastAPI()
call_count = {"n": 0}  # 演示用计数器


def expensive_dep() -> int:
    call_count["n"] += 1
    return 42


def dep_a(v: Annotated[int, Depends(expensive_dep)]) -> int:
    return v * 2


def dep_b(v: Annotated[int, Depends(expensive_dep)]) -> int:
    return v + 1


@app.get("/cached/")
async def cached(
    a: Annotated[int, Depends(dep_a)],
    b: Annotated[int, Depends(dep_b)],
    raw: Annotated[int, Depends(expensive_dep)],
):
    # expensive_dep 在整个请求中只执行 1 次,三个位置拿到同一个值
    return {"a": a, "b": b, "raw": raw, "calls": call_count["n"]}

访问 /cached/ 会看到 calls 恒为该请求内的 1(每个请求独立计数)。这个特性对"校验当前用户"这类依赖至关重要:认证依赖可能被安全链、权限链、路径函数多处引用,缓存保证每请求只查一次数据库。

如果确实需要每次都重新执行(例如每次生成新的时间戳/随机值),显式禁用缓存:

python
def timestamp_dep(ts: Annotated[float, Depends(get_time, use_cache=False)]) -> float:
    return ts

缓存范围是"单次请求"

缓存不是全局缓存:不同请求之间互不共享,每个请求都会重新执行依赖。需要跨请求复用请用 lifespan + 模块级单例(第 21 章)。

10.6 本章小结

  • 依赖 = 参数与路径函数同构的普通函数(callable),用 Depends(fn) 声明,不加括号;
  • 依赖声明的参数同样会被解析、校验并写入 OpenAPI 文档;
  • Annotated[X, Depends(fn)] 可存为类型别名复用,是官方推荐写法,类型检查完整保留;
  • async defdef 依赖可任意混用,FastAPI 自动选择执行方式;
  • 同一请求内同一依赖默认只执行一次(use_cache 控制),缓存范围是单请求。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 声明依赖 commons: Annotated[dict, Depends(common_parameters)] 时,正确写法是?

2. 官方推荐用 Annotated[dict, Depends(fn)] 而不是 dict = Depends(fn) 的核心理由是?

3. 一个请求中,依赖 A 和依赖 B 都声明了同一个子依赖 expensive_dep,默认会发生什么?

4. 关于同步与异步依赖,下列说法正确的是?

🛠️ 动手实践

  1. 把第 3 章的分页参数(skip/limit)重构成 Pagination 依赖并用类型别名在三个接口上复用,观察 /docs 中的参数展示。
  2. 写一个 get_db 依赖返回模拟的数据库会话对象,验证两个依赖 + 路径函数同时引用它时,一个请求内会话对象是同一个(用 id(session) 对比)。
  3. 实现一个 request_id 依赖:use_cache=False,每次调用生成新的 UUID,并在两个依赖中同时引用,打印日志观察它被执行了两次。

基础打好后,下一章把依赖注入推向生产级用法:第 11 章 · 依赖注入进阶