📚 编程进阶教程站

第 11 章 · 依赖注入进阶 ​

本章目标：掌握类作为依赖、多层子依赖树、dependencies=[...] 副作用式依赖、全局依赖与 yield 依赖（含异常处理的坑），达到生产级 DI 使用水平。

11.1 类作为依赖 ​

任何可调用对象（callable）都能当依赖——函数是，类也是。创建实例 Cat(name="Mr Fluffy") 本质就是"调用"类，所以 FastAPI 会分析类的 __init__ 参数并按路径函数参数同样的方式解析：

python
from typing import Annotated, Optional
from fastapi import FastAPI, Depends

app = FastAPI()


class CommonQueryParams:
    """__init__ 的参数就是依赖声明的请求参数"""

    def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit


@app.get("/items/")
async def list_items(
    commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)],
):
    # 拿到的是实例：属性有类型、可补全，比裸 dict 友好得多
    return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

注意一个细节：Annotated[CommonQueryParams, Depends(CommonQueryParams)] 里类名出现了两次。真正起作用的是 Depends(...) 里的那个；类型标注只是给编辑器看的。

当两者相同时可以写快捷方式 Depends() 不带参数：

python
@app.get("/items-short/")
async def list_items_short(
    commons: Annotated[CommonQueryParams, Depends()],  # FastAPI 从类型标注推断
):
    return {"skip": commons.skip}

相比返回 dict，类依赖的价值在于：属性可自动补全、便于写单元测试（直接构造实例）、还能在类里挂方法承载行为。

11.2 子依赖树：任意深度的嵌套 ​

依赖自己也可以声明依赖，形成一棵任意深的依赖树，FastAPI 负责自底向上逐层求解：

python
from typing import Annotated, Optional
from fastapi import Cookie, Depends, FastAPI

app = FastAPI()


def query_extractor(q: Optional[str] = None):
    return q


def query_or_cookie_extractor(
    q: Annotated[str, Depends(query_extractor)],  # 依赖里再声明依赖
    last_query: Annotated[Optional[str], Cookie()] = None,
):
    if not q:
        return last_query   # 查询串没有就退回上次保存在 cookie 的查询
    return q


@app.get("/items/")
async def read_query(
    result: Annotated[str, Depends(query_or_cookie_extractor)],
):
    return {"q_or_last": result}

你只在路径函数上声明了一个依赖，但 FastAPI 知道必须先解出 query_extractor 再调用外层。安全体系正是建立在这之上：当前用户 → 活跃用户 → 管理员用户 → 接口 这样的权限链。

缓存与"只执行一次"

第 10 章的请求级缓存在依赖树中同样生效：树中多个分支引用同一叶子时，该叶子每请求只求值一次。

11.3 dependencies=[...]：只要副作用，不要返回值 ​

有些场景只需要依赖被执行（如校验密钥），并不关心它的返回值。这时不必在路径函数签名里加参数，直接在装饰器上传 dependencies 列表：

python
from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


async def verify_key(x_key: str = Header(...)):
    if x_key != "preset-secret":
        raise HTTPException(status_code=400, detail="X-Key 无效")


async def verify_token(x_token: str = Header(...)):
    if x_token != "preset-token":
        raise HTTPException(status_code=401, detail="X-Token 无效")
    # 有返回值也没关系，不会传给路径函数


@app.get("/protected/", dependencies=[Depends(verify_key), Depends(verify_token)])
async def protected():
    return {"data": "机密数据"}

这种写法的三个好处：

  • 编辑器不会提示"未使用的参数"；
  • 权限逻辑与业务函数彻底解耦，新增接口时一行装饰器即可复用整套检查；
  • 依赖内部照样能抛 HTTPException 中断请求。

11.4 全局依赖与路由组依赖 ​

把 dependencies 提升到应用级别，则所有路径操作都要先过这些检查：

python
from fastapi import Depends, FastAPI

app = FastAPI(dependencies=[Depends(log_request)])  # 全局：每个接口都执行


@app.get("/open/")      # 也受全局依赖约束
async def open_route():
    return {"ok": True}

更精细的分组方式要等第 19 章 APIRouter(prefix="/admin", dependencies=[...])：给一组接口统一加权限，而不影响公开接口。

11.5 yield 依赖：带清理逻辑的资源管理 ​

数据库会话这类资源需要"用完关闭"。把 return 换成 yield，yield 之前的代码在响应生成前执行，之后的代码在响应发送后执行：

python
from fastapi import Depends, FastAPI

app = FastAPI()


async def get_db():
    db = SessionLocal()          # ① 前置：创建会话
    try:
        yield db                 # ② 注入给路径函数使用
    finally:
        await db.close()         # ③ 后置：无论成败都执行清理


@app.get("/users/{user_id}")
async def get_user(user_id: int, db=Depends(get_db)):
    return db.query(User).get(user_id)

它本质上就是一个 @contextlib.asynccontextmanager 风格的上下文管理器——FastAPI 内部正是这样实现的。

异常处理的两个铁律 ​

python
async def risky_db():
    session = SessionLocal()
    try:
        yield session
    except HTTPException:
        # 可以捕获路径函数冒泡上来的异常，甚至换成新的 HTTPException
        await session.rollback()
        raise HTTPException(status_code=400, detail="事务已回滚")
    finally:
        await session.close()
  1. except 之后必须重新 raise：如果在 yield 依赖中吞掉异常既不抛原异常也不抛新异常，客户端仍会收到 500，但服务端没有任何日志——这是官方文档专门警告的隐蔽事故源；
  2. 多层 yield 子依赖的退出顺序与进入顺序相反（栈式），FastAPI 保证每个依赖的后置代码按正确顺序执行，因此深层依赖的清理代码仍能访问其上游 yield 出来的值。

11.6 本章小结 ​

  • 类是 callable，__init__ 参数即依赖声明；类型与 Depends() 相同时可用 Depends() 快捷写法；
  • 子依赖可无限嵌套，缓存机制保证同一叶子每请求只求值一次；
  • dependencies=[Depends(fn)] 用于纯副作用场景（鉴权/限流），返回值被丢弃；
  • 全局依赖挂在 FastAPI(dependencies=[...])，分组依赖挂在 APIRouter；
  • yield 依赖 = 前置 + 注入 + 后置三段式；except 后必须重新 raise，否则异常会被静默吞掉且无日志。

🧪 随堂测验

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

1. commons: Annotated[CommonQueryParams, Depends()] 这个快捷写法成立的前提是？

2. dependencies=[Depends(verify_token)] 与参数声明 Depends 的核心区别是？

3. 在 yield 依赖中 except 捕获了异常却既不重新 raise 也不抛新异常，后果是？

4. 关于 yield 依赖的执行时机，正确的是？

🛠️ 动手实践 ​

  1. 把第 10 章实践 2 的 get_db 改造成 yield 依赖，并在其中加入 try/finally 与回滚逻辑；故意在路径函数中 raise 一个异常验证清理代码仍被执行。
  2. 实现 require_admin(x_admin_key: str = Header(...)) 依赖，通过全局 dependencies=[...] 保护 /admin/* 下三个接口，并用 curl 验证缺 key 时返回 401。
  3. 构建三层子依赖链 get_settings → get_db(settings) → get_current_user(db)，在路径函数中只声明最上层依赖，打印三层对象的创建顺序日志验证求解顺序。

依赖注入两大章收官。下一章学习横切关注点的另一种形态：第 12 章 · 中间件与 CORS。