第 6 章：依赖注入系统

鉴权、数据库会话、分页参数、限流——这些逻辑会出现在几乎每个处理函数里。依赖注入（Dependency Injection）是 FastAPI 处理这类横切关注点的统一机制，也是它区别于 Flask 的核心设计。

学习目标

理解依赖注入要解决的问题，以及它为什么让复用与测试变简单
掌握 Depends()、类依赖、子依赖与依赖树的解析顺序
说清同一次请求内依赖结果的缓存语义，以及何时必须 use_cache=False
能用 yield 依赖正确管理数据库会话、临时文件等资源
掌握依赖的三种挂载层级，并用 Annotated 统一依赖声明

📌 本章示例仍为单文件；第 10、11 章会把 get_db 换成真正的 AsyncSession，第 12 章会在 get_current_user 里做 JWT 校验。

6.1 依赖注入解决什么问题

不用依赖注入时，一个带鉴权和分页的接口长这样：

python

@app.get("/articles")
async def list_articles(request: Request, page: int = 1, size: int = 20):
    # 1. 手工解析 token
    token = request.headers.get("Authorization", "").removeprefix("Bearer ")
    if not token:
        raise HTTPException(status_code=401, detail="未登录")
    # 2. 手工查用户
    user = await lookup_user(token)
    if user is None:
        raise HTTPException(status_code=401, detail="无效凭证")
    # 3. 手工开数据库会话
    session = SessionLocal()
    try:
        ...  # 4. 业务逻辑终于开始了——而这段在别的接口里还要重抄一遍
    finally:
        session.close()

问题很明确：噪音淹没了业务，而且每一份拷贝都可能漏掉 finally、漏掉权限判断。依赖注入把这些「参数解析」抽走，交给框架在处理函数被调用之前完成：

python

@app.get("/articles")
async def list_articles(
    user: CurrentUser,                              # 鉴权：谁在调用
    session: DbSession,                             # 资源：数据库会话
    pagination: Annotated[Pagination, Depends()],   # 参数：分页
) -> list[ArticleOut]:
    ...                                             # 只剩业务

维度	收益
复用	一个 `get_current_user` 服务所有需要登录的端点，改一处全站生效
测试	`app.dependency_overrides` 能替换任意依赖，不必真连数据库（见第 16 章）
文档	依赖里声明的 header / query 参数会自动出现在 OpenAPI 里
分层	依赖可以互相依赖，自然形成 `db → repository → service` 的调用链

6.2 最基础的 `Depends`

python

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class CommonQuery:
    def __init__(self, q: str | None = None, limit: int = 20, offset: int = 0) -> None:
        self.q = q
        self.limit = min(limit, 100)  # 防止客户端传超大 limit
        self.offset = offset


async def common_parameters(q: str | None = None, limit: int = 20, offset: int = 0) -> CommonQuery:
    return CommonQuery(q=q, limit=limit, offset=offset)


# 推荐写法：Annotated，调用处简洁且可复用
CommonParams = Annotated[CommonQuery, Depends(common_parameters)]


@app.get("/articles")
async def list_articles(params: CommonParams) -> dict[str, object]:
    return {"q": params.q, "limit": params.limit, "offset": params.offset}


@app.get("/users")
async def list_users(params: CommonParams) -> dict[str, object]:
    # 同一个依赖，零重复代码
    return {"q": params.q, "limit": params.limit, "offset": params.offset}

关键点：

依赖函数的参数由 FastAPI 解析。common_parameters 声明了三个标量参数，FastAPI 会把它们当作查询参数处理，并自动写进两个端点的 OpenAPI 文档。
依赖可以是 def 也可以是 async def。同步依赖会被丢进线程池执行，不阻塞事件循环（第 13 章详述）。
Depends(common_parameters) 传的是可调用对象本身，不是调用结果。写成 Depends(common_parameters()) 会立刻执行函数并把它当依赖，是典型错误。

💡 Annotated[CommonQuery, Depends(common_parameters)] 与 params: CommonQuery = Depends(common_parameters) 等价。本书统一用 Annotated：可作类型别名复用，也不与默认值冲突。

6.3 类作为依赖

把类本身传给 Depends，FastAPI 会用它 __init__ 声明的参数作为依赖参数：

python

from fastapi import Depends, FastAPI, HTTPException, status

app = FastAPI()


class Pagination:
    def __init__(self, page: int = 1, size: int = 20) -> None:
        if page < 1 or size < 1:
            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail="page 与 size 必须为正整数",
            )
        self.page = page
        self.size = min(size, 100)

    @property
    def offset(self) -> int:
        return (self.page - 1) * self.size


@app.get("/items")
async def list_items(pagination: Annotated[Pagination, Depends()]) -> dict[str, int]:
    return {"page": pagination.page, "size": pagination.size, "offset": pagination.offset}

也可以显式写 Depends(Pagination)，但 Annotated[Pagination, Depends()] 更省事。

为什么用类而不是函数：当依赖需要携带配置、靠参数派生出多个变体时，类更自然。写法是让实例实现 __call__：

python

class RateLimiter:
    def __init__(self, times: int, seconds: int) -> None:
        self.times = times
        self.seconds = seconds

    async def __call__(self, request: Request, response: Response) -> None:
        # 真实实现用 Redis：SETNX key 计数 + EXPIRE
        response.headers["X-RateLimit-Limit"] = str(self.times)


# 同一个类，两份配置
strict_limiter = RateLimiter(times=5, seconds=60)


@app.post("/auth/login", dependencies=[Depends(strict_limiter)])
async def login() -> dict[str, str]:
    return {"msg": "ok"}

类实例通过 __call__ 变成依赖函数，配置则在构造时注入。 这是「有配置的依赖」的标准写法，不必为每种配置单独定义函数。

6.4 子依赖与依赖树

依赖可以再依赖别的依赖，FastAPI 会递归解析成一棵树，自底向上构建：

python

class FakeSession:
    def close(self) -> None:
        print("[cleanup] session closed")


def get_settings() -> dict[str, str]:
    return {"db_url": "postgresql+asyncpg://localhost/app", "env": "dev"}


def get_db(settings: Annotated[dict[str, str], Depends(get_settings)]) -> FakeSession:
    # 真实实现：create_async_engine(...) + async_sessionmaker
    return FakeSession()


class UserRepository:
    def __init__(self, session: FakeSession) -> None:
        self.session = session

    async def get_by_username(self, username: str) -> dict[str, object] | None:
        if username != "moqian":
            return None
        return {"id": 1, "username": "moqian", "password_hash": "$argon2id$..."}


def get_user_repository(session: Annotated[FakeSession, Depends(get_db)]) -> UserRepository:
    return UserRepository(session)


class UserService:
    def __init__(self, repo: UserRepository) -> None:
        self.repo = repo

    async def authenticate(self, username: str) -> dict[str, object] | None:
        return await self.repo.get_by_username(username)


def get_user_service(repo: Annotated[UserRepository, Depends(get_user_repository)]) -> UserService:
    return UserService(repo)

两个容易忽略的性质：

同一层级的多个依赖按参数声明顺序解析，重复依赖只解析一次（见 6.5）。
清理严格逆序：get_db 中 yield 之后的代码在响应发出后才执行，晚于处理函数、晚于所有子依赖的清理。

6.5 依赖结果缓存：`use_cache` 的语义

默认情况下，同一个依赖在一次请求内只执行一次，结果被缓存复用，这就是 use_cache=True（默认值）的含义——缓存范围是单次请求，不是全局。

python

calls: list[str] = []


def get_trace_id() -> str:
    calls.append("executed")
    return f"trace-{len(calls)}"


CacheDep = Annotated[str, Depends(get_trace_id)]
NoCacheDep = Annotated[str, Depends(get_trace_id, use_cache=False)]


@app.get("/cached")
async def cached(a: CacheDep, b: CacheDep) -> dict[str, object]:
    # 两个参数拿到同一个值，get_trace_id 只执行一次
    return {"a": a, "b": b, "same": a is b}


@app.get("/uncached")
async def uncached(a: NoCacheDep, b: NoCacheDep) -> dict[str, object]:
    # 每个参数各执行一次，拿到不同的值
    return {"a": a, "b": b}

实测输出：/cached 返回 {"a": "trace-1", "b": "trace-1", "same": true}，而 /uncached 返回 {"a": "trace-2", "b": "trace-3"}。

场景	用 `use_cache`	原因
数据库会话、当前用户、配置	`True`（默认）	一次请求共享一个连接/身份，避免重复查询
生成一次性随机数、签名 nonce	`False`	每次调用都必须产生新值
每次都要重新读时间的审计戳	`False`	缓存会让同一请求内多次取值完全相同
高开销但结果只读的资源	`True`	缓存正是为了省这次开销

⚠️ 依赖的缓存键是「依赖可调用对象 + 安全参数」，不包含依赖函数内部读到的状态。若依赖读数据库或全局变量，缓存的含义是「本次请求内看到的是第一次调用时的快照」。

6.6 `yield` 依赖：资源管理

yield 依赖把「获取资源」和「释放资源」写在一个函数里，FastAPI 保证释放代码一定被执行。

python

from collections.abc import AsyncGenerator
from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FakeSession:
    def close(self) -> None:
        print("[cleanup] session closed")


async def session_commit(session: FakeSession) -> None:
    print("[cleanup] commit on success")


async def session_rollback(session: FakeSession) -> None:
    print("[cleanup] rollback on error")


async def get_db() -> AsyncGenerator[FakeSession, None]:
    session = FakeSession()
    print("[setup] session created")
    try:
        yield session                      # ← 处理函数在这里拿到 session
        await session_commit(session)
    except Exception:
        await session_rollback(session)
        raise                              # 必须重新抛出，否则异常被吞掉
    finally:
        session.close()                    # ← 无论成功失败都会执行


@app.get("/users")
async def list_users(session: Annotated[FakeSession, Depends(get_db)]) -> list[str]:
    return ["moqian"]

执行时序（最容易理解错的地方）：

阶段	发生了什么
1	依赖函数执行到 `yield`，把值交给 FastAPI
2	处理函数运行，业务逻辑与响应模型序列化完成
3	响应已经发给客户端（HTTP 层面已完成）
4	FastAPI 恢复依赖函数，执行 `yield` 之后的代码：提交 / 回滚 / 关闭

由此推出两条规则：不要指望在 yield 之后修改响应（响应早已发出）；try / finally 不是可选项。只写 yield 然后跟在后面的清理代码，一旦处理函数抛异常就永远不会执行：

❌ 不推荐：

python

async def get_db_bad() -> AsyncGenerator[FakeSession, None]:
    session = FakeSession()
    yield session
    session.close()  # 处理函数抛异常时这行被跳过，连接泄漏

✅ 推荐：

python

async def get_db_good() -> AsyncGenerator[FakeSession, None]:
    session = FakeSession()
    try:
        yield session
        await session_commit(session)
    except Exception:
        await session_rollback(session)
        raise
    finally:
        session.close()

实测中，正常请求打印 setup → commit → close，处理函数抛 HTTPException 时打印 setup → rollback → close。这正是第 10 章数据库会话依赖的写法。

💡 yield 依赖可以是同步函数（Generator）也可以是异步函数（AsyncGenerator）。同步版本在线程池里运行，用它包阻塞 IO 是合理的。

6.7 依赖的三种挂载层级

除了「作为参数注入」，依赖还能挂在三层容器上，用于必须先执行、但不需要返回值的场景（典型是鉴权）：

python

from fastapi import APIRouter, Depends, FastAPI, HTTPException, status

async def verify_api_key(x_api_key: str | None = None) -> None:
    if x_api_key != "secret-key":
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="API Key 无效")


async def log_request() -> None:
    print("[access] request received")


# 层级 3：全局，所有端点都执行
app = FastAPI(dependencies=[Depends(log_request)])

# 层级 2：路由器级，该 router 下所有端点执行
admin_router = APIRouter(prefix="/admin", dependencies=[Depends(verify_api_key)])

# 层级 1：路径操作级，只对这一个端点执行
@app.get("/health", dependencies=[Depends(verify_api_key)])
async def health() -> dict[str, str]:
    return {"status": "ok"}

层级	写法	作用范围	典型用途
1	`@app.get(..., dependencies=[...])`	单个端点	单点鉴权、审计
2	`APIRouter(dependencies=[...])`	该 router 全部端点	`/admin/*` 统一验管理员
3	`FastAPI(dependencies=[...])`	全应用	全局访问日志、请求 ID 注入

三层依赖都按「外到内」的顺序执行，任何一层抛 HTTPException 都会中止后续解析。

关键限制：这三个层级都不接收依赖的返回值。 你只能在依赖内部通过 raise HTTPException 表达失败，无法拿到返回的对象参与业务计算。需要返回值就写成处理函数的参数。

⚠️ 全局依赖会被每个端点执行，包括 /docs、/openapi.json。别在其中做重活（如查数据库），否则文档页也会变慢。

6.8 `Annotated` 统一依赖声明

当依赖在几十个端点里反复出现时，每次都写 Depends(get_current_user) 很啰嗦。标准做法是抽成类型别名：

python

from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException, status

app = FastAPI()


class CurrentUserModel:
    def __init__(self, id: int, username: str, scopes: set[str]) -> None:
        self.id = id
        self.username = username
        self.scopes = scopes


async def get_current_user(
    authorization: Annotated[str | None, Header()] = None,
) -> CurrentUserModel:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="未提供有效的 Bearer Token",
            headers={"WWW-Authenticate": "Bearer"},  # RFC 6750 要求
        )
    token = authorization.removeprefix("Bearer ")
    # 真实实现：PyJWT 解码 + 查库，见第 12 章
    if token != "demo-token":
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token 无效")
    return CurrentUserModel(id=1, username="moqian", scopes={"articles:read"})


# 一行别名，全项目复用
CurrentUser = Annotated[CurrentUserModel, Depends(get_current_user)]


@app.get("/me")
async def read_me(user: CurrentUser) -> dict[str, object]:
    return {"id": user.id, "username": user.username}

三个好处：调用处只看到业务语义（user: CurrentUser）；别名可以集中升级（第 12 章换真正的 JWT 实现只改一处）；IDE 知道 user 是 CurrentUserModel，能补全 .username。

💡 依赖内部声明参数时 Header / Cookie / Query / Path 都能用。注意 Path 参数要求路径模板中存在同名变量，否则报错。

6.9 测试中的依赖覆盖

依赖注入最大的附加价值是可测试性。app.dependency_overrides 是一个字典，把「原依赖」映射到「替身」：

python

from fastapi.testclient import TestClient

client = TestClient(app)


def fake_current_user() -> CurrentUserModel:
    return CurrentUserModel(id=999, username="tester", scopes={"articles:write"})


app.dependency_overrides[get_current_user] = fake_current_user
try:
    resp = client.get("/me")
    assert resp.json() == {"id": 999, "username": "tester"}
finally:
    app.dependency_overrides.clear()  # 必须清理，否则污染其他测试

要点：

键必须是原始依赖对象本身（get_current_user），不是 Annotated 别名，也不是调用结果。
覆盖会沿着依赖树生效：替换 get_db 后，所有依赖它的 repository / service 都会拿到测试会话。第 16 章的测试里只覆盖最底层的 get_db 就够。
dependency_overrides 是应用级状态，测试间必须清理；放在 pytest fixture 里用 yield 收尾最稳。

常见坑与排查

现象	原因	解决
数据库连接越用越多，最终耗尽	`yield` 依赖的清理代码直接写在 `yield` 之后，没包 `try/finally`	用 `try: yield x; commit() except: rollback(); raise finally: close()`
处理函数抛异常，`yield` 后的 commit 仍执行了	没写 `except` 分支，收尾只有 commit	加 `except` 分支做 rollback 并 `raise` 重新抛出
同一个依赖在一次请求里执行了多次	某个参数位置用了 `use_cache=False`，或误以为缓存是全局的	需要共享结果就保留默认 `use_cache=True`；确认缓存范围是单次请求
全局/路由器级依赖的返回值拿不到	`dependencies=[...]` 只执行不传值，这是设计如此	需要值就改成处理函数参数 `user: CurrentUser`
依赖写成了普通函数，直接在函数体里 `await get_current_user()` 调用	绕过了注入，参数解析、文档、缓存、覆盖全部失效	一律通过 `Depends()` / `Annotated` 声明
启动或请求时报 `Path parameter ... not found`	依赖内部声明了 `Path` 参数，但路径模板里没有同名变量	路径变量名与依赖参数名保持一致，或改用 `Query`
两个依赖互相 `Depends`，报递归或启动失败	循环依赖，依赖树无法拓扑排序	抽出公共的第三方依赖（如 `get_db`），让两者都依赖它
依赖里 `return None` 表示「未登录」，接口仍返回 200	返回 `None` 是合法返回值，不构成失败	失败必须 `raise HTTPException`
测试之间互相影响，鉴权行为随机	忘记清理 `app.dependency_overrides`	放进 pytest fixture 的 `yield` 之后 `clear()`

本章小结

要点	说明
核心目的	把参数解析与资源管理从业务逻辑中剥离，换来复用、可测与统一文档
基本形式	`Depends(callable)`；用 `Annotated` 抽别名是本书统一风格
类依赖	类实例实现 `__call__` 后可作为依赖，构造参数用于携带配置
子依赖	依赖可嵌套成树，自底向上解析，清理严格逆序
缓存	默认 `use_cache=True`，缓存范围是单次请求；需每次重算时关掉
`yield` 依赖	`try / yield / finally` 管理会话、文件、锁；收尾在响应发出后执行
挂载层级	路径操作 / `APIRouter` / `FastAPI` 三级，都不接收返回值
依赖参数	依赖函数可声明 query / header / cookie / path 参数，由 FastAPI 一并解析
测试覆盖	`app.dependency_overrides[原依赖] = 替身`，用完必须清理
失败表达	依赖里失败一律 `raise HTTPException`，不要用返回 `None` 表示失败

练习题

写一个依赖工厂 paginate(max_size: int = 100)，返回一个依赖函数，把 page / size 解析成带 offset 属性的对象并强制 size <= max_size。用它在 GET /articles 与 GET /comments 上，写出 Annotated 别名。
实现 get_db 的 yield 依赖：进入时打印 setup、正常结束时打印 commit、异常时打印 rollback 并重新抛出、最后总是打印 close。分别用「正常返回」和「处理函数抛 HTTPException」两种请求验证输出顺序。
给定 get_db、get_repo（依赖 get_db）、get_service（依赖 get_repo）三个依赖，画出依赖树 mermaid 图，标出各节点执行顺序、缓存生效的位置、清理发生的时机。
用 app.dependency_overrides 写一个测试：把 get_current_user 替换成固定用户，请求 /me 并断言返回该用户的 id 与 username；然后用 try/finally 清理覆盖，说明不清理会造成什么后果。
下面代码有三处问题，请指出并改正：

python

async def get_session():
    session = SessionLocal()
    yield session
    session.close()


@app.get("/items")
async def list_items(session: Annotated[Session, Depends(get_session)]):
    if not session.is_active:
        return None
    return session.query(Item).all()

下一章预告

依赖解决了「怎么拿参数」，但参数本身的花样还没讲完：Path / Query / Body / Header / Cookie / Form / File 的完整约束，以及它们与 Pydantic 模型的组合边界。
👉 第 7 章：参数进阶

第 6 章：依赖注入系统 ​

学习目标 ​

6.1 依赖注入解决什么问题 ​

6.2 最基础的 Depends ​

6.3 类作为依赖 ​

6.4 子依赖与依赖树 ​

6.5 依赖结果缓存：use_cache 的语义 ​

6.6 yield 依赖：资源管理 ​

6.7 依赖的三种挂载层级 ​

6.8 Annotated 统一依赖声明 ​

6.9 测试中的依赖覆盖 ​

常见坑与排查 ​

本章小结 ​

练习题 ​

下一章预告 ​

第 6 章：依赖注入系统

学习目标

6.1 依赖注入解决什么问题

6.2 最基础的 `Depends`

6.3 类作为依赖

6.4 子依赖与依赖树

6.5 依赖结果缓存：`use_cache` 的语义

6.6 `yield` 依赖：资源管理

6.7 依赖的三种挂载层级

6.8 `Annotated` 统一依赖声明

6.9 测试中的依赖覆盖

常见坑与排查

本章小结

练习题

下一章预告