FastAPI 依赖注入：把请求前置工作只写一次

一个接口真正开始处理业务之前，往往已经有一串事情要做：读查询参数、检查请求头、识别当前用户、判断权限、打开数据库会话。刚开始写项目时，把这些代码塞进每个接口似乎最直接。等接口从三个长到三十个，问题就冒出来了：同一段鉴权被复制很多遍，某个接口忘了补新规则，数据库会话遇到异常时没有关闭，测试还得真的去碰外部服务。

FastAPI 的依赖注入，解决的就是这类重复的请求前置工作。你把一项工作写成普通函数，再用 Depends 声明“这个接口运行前需要它”。请求到来后，FastAPI 会读取函数签名，准备参数，按依赖关系调用函数，把返回值交给下一层，最后才调用接口函数。

这章不从抽象的设计模式讲起。我们会从一个能发请求的商品接口开始，逐步加上嵌套依赖、认证授权、数据库会话、请求内缓存和测试覆盖。每一处都盯着一个问题：代码运行起来时，FastAPI 到底先做了什么，又把什么值交给了谁。

先把重复的参数处理拿出来

假设商品列表和订单列表都支持关键字、偏移量和每页数量。最容易想到的写法，是在两个接口函数里各写一遍参数：

@app.get("/products")
async def list_products(
    keyword: str | None = None,
    offset: int = 0,
    size: int = 10,
):
    ...
 
 
@app.get("/orders")
async def list_orders(
    keyword: str | None = None,
    offset: int = 0,
    size: int = 10,
):
    ...

代码暂时不长，可约束很快会变多。比如 offset 不能小于零，size 只能在 1 到 50 之间，关键字最多 20 个字符。复制的不再只是三个参数，而是三套校验规则。以后把每页上限改成 100 时，还要确认所有接口有没有一起改。

我们先把这组规则写成一个依赖函数：

from dataclasses import dataclass
from typing import Annotated
 
from fastapi import Depends, FastAPI, Query
 
app = FastAPI()
 
 
@dataclass(frozen=True)
class PageWindow:
    keyword: str | None
    offset: int
    size: int
 
 
def page_window(
    keyword: Annotated[str

page_window 没有路由装饰器，看起来就是一个普通 Python 函数。它的参数却可以像接口参数一样使用 Query、默认值和类型标注。FastAPI 处理请求时，会用同一套规则为它提取和校验数据。

依赖函数返回的 PageWindow，就是下一步要注入的值：

PageWindowDep = Annotated[PageWindow, Depends(page_window)]
 
PRODUCTS = ["键盘", "鼠标", "显示器", "扩展坞", "机械键盘"]
 
 
@app.get("/products")
async def list_products(window: PageWindowDep):
    selected = [
        item
        for item in PRODUCTS
        if not window.keyword or window.keyword in item

发送下面的请求：

GET /products?keyword=键盘&offset=0&size=2

接口返回：

{
  "items": ["键盘", "机械键盘"],
  "offset": 0,
  "size": 2
}

这里发生的事情可以按时间拆成四步：

如果把 size 改成 0，接口函数根本不会执行。校验在依赖阶段就失败，请求得到 422：

{
  "detail": [
    {
      "type": "greater_than_equal",
      "loc": ["query", "size"],
      "msg": "Input should be greater than or equal to 1",
      "input": "0",
      "ctx": {"ge": 1}
    }
  ]
}

依赖声明的请求参数也会进入 OpenAPI。打开接口文档时，keyword、offset、size 仍然会显示在 /products 下面。把参数抽进依赖，并不会让文档丢失这部分信息。

读懂 Depends 和 Annotated

第一次见到下面这行代码，最容易被两层括号绕住：

window: Annotated[PageWindow, Depends(page_window)]

可以把它从里往外读：

• PageWindow 告诉编辑器和类型检查工具，window 在函数体内是什么类型。
• Depends(page_window) 告诉 FastAPI，这个值应该通过调用 page_window 得到。
• Annotated 把“值的 Python 类型”和“框架要读取的额外信息”放在同一个标注里。

Python 本身仍然把这个参数视为 PageWindow。FastAPI 运行时会额外读取 Depends(page_window) 这份元数据。这样写以后，接口函数里的自动补全仍然知道 window.keyword、window.offset 和 window.size。

传函数，不要先调用函数

Depends 接收的是一个可调用对象。这里传 page_window，不要写成 page_window()：

# 正确：把函数交给 FastAPI
Depends(page_window)
 
# 错误方向：应用导入时就调用了函数
Depends(page_window())

请求参数只有请求到来以后才存在。FastAPI 必须拿到函数本身，才能在每次请求中读取它的签名、准备实参并决定什么时候调用。

注入依据不是返回类型

FastAPI 不会因为参数类型写了 PageWindow，就自动在全局寻找一个能返回 PageWindow 的函数。真正指出依赖来源的是 Depends(page_window)。

这意味着两个依赖都可以返回 User，但执行完全不同的流程：一个从访问令牌查用户，另一个从内部任务上下文取用户。你在哪个参数上写了哪个 Depends，FastAPI 就调用哪个可调用对象。它不是“按类型自动装配”。

用类型别名收起重复声明

同一个依赖会被很多接口使用时，可以把整段 Annotated 保存成类型别名：

PageWindowDep = Annotated[PageWindow, Depends(page_window)]
 
 
@app.get("/products")
async def list_products(window: PageWindowDep):
    ...
 
 
@app.get("/orders")
async def list_orders(window: PageWindowDep):
    ...

这个别名没有提前执行依赖，也没有创建共享对象。它只是把同一份类型和依赖元数据起了一个短名字。每次请求到来时，FastAPI 仍然会正常解析 page_window。

只执行，不接收返回值

有些依赖只负责检查，例如确认内部请求头存在。接口函数并不需要它的返回值，可以把依赖放到路由装饰器上：

from typing import Annotated
 
from fastapi import Header, HTTPException
 
 
def verify_internal_key(
    x_internal_key: Annotated[str | None, Header()] = None,
) -> None:
    if x_internal_key != "demo-key":
        raise HTTPException(status_code=403, detail="内部密钥无效")
 
 
@app.get(

依赖照样先执行，失败时照样中止请求，只是接口签名里不会多出一个用不到的参数。相同方式还能放到 APIRouter 上保护整组路由，或放到 FastAPI 实例上作用于整个应用。

依赖会继续依赖

真实项目里的前置工作很少只有一层。权限检查需要当前用户，当前用户来自访问令牌；订单查询还需要数据库会话。把这些工作拆成小依赖以后，它们会连成一张有方向的图。

我们为订单接口加一条认证链。先定义用户模型和令牌提取器：

from typing import Annotated
 
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel
 
 
class User(BaseModel):
    username: str
    roles: list[str]
 
 
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
TokenDep = Annotated[str, Depends(oauth2_scheme)]

OAuth2PasswordBearer 是一个可调用对象，所以也能交给 Depends。请求头是 Authorization: Bearer alice-token 时，它会提取出 alice-token 这个字符串。缺少 Bearer 令牌时，请求会在这里得到 401，后面的用户查询和接口函数都不会执行。

下一层用令牌查当前用户：

async def get_current_user(token: TokenDep) -> User:
    users = {
        "alice-token": User(
            username="alice",
            roles=["reader", "editor"],
        ),
        "bob-token": User(
            username="bob",
            roles=["reader"],
        ),
    }
    if token not in users:

get_current_user 自己就是依赖，同时它又依赖 oauth2_scheme。FastAPI 会先解决更深的令牌依赖，再把令牌字符串传给 get_current_user。

权限检查再向上一层：

def require_editor(user: CurrentUserDep) -> User:
    if "editor" not in user.roles:
        raise HTTPException(status_code=403, detail="需要编辑权限")
    return user
 
 
EditorDep = Annotated[User, Depends(require_editor)]

订单接口只声明自己最终需要什么：

@app.get("/orders/{order_id}")
async def read_order(
    order_id: int,
    editor: EditorDep,
    session: SessionDep,
):
    order = session.load_order(order_id)
    return {
        "operator": editor.username,
        "order": order,
        "session_id": session.session_id,
    }

请求带上 Alice 的令牌：

GET /orders/17
Authorization: Bearer alice-token

返回内容是：

{
  "operator": "alice",
  "order": {
    "id": 17,
    "status": "待发货"
  },
  "session_id": 1
}

接口函数没有解析请求头，没有验证令牌，也没有自己判断角色。它拿到的 editor 已经是通过检查的 User，拿到的 session 已经是可以查询的会话。

依赖失败会让后面的工作短路

Bob 的令牌能识别出用户，但 Bob 没有 editor 角色：

GET /orders/17
Authorization: Bearer bob-token

返回：

{
  "detail": "需要编辑权限"
}

状态码是 403。read_order 不会运行。按照上面接口参数的依赖顺序，数据库会话也还没有打开。这一点很有工程价值：把便宜的格式检查、认证和权限检查放在昂贵资源之前，无权访问的请求就不会占用数据库连接。

图不是简单的调用链

如果两个上层依赖都需要 get_current_user，依赖关系会出现分叉和汇合。FastAPI 会解析整张图，而不是把每个参数当成互不相关的函数调用。默认情况下，同一请求内重复出现的同一个依赖会复用计算结果。后面讲 use_cache 时，我们会直接看到这个行为。

把认证和授权分成两道门

认证回答“请求者是谁”，授权回答“这个人能不能做这件事”。它们经常连在一起，却不应该混成一个巨大函数。

在刚才的代码里：

• oauth2_scheme 只负责从标准位置取得 Bearer 令牌。
• get_current_user 验证令牌并返回用户，失败使用 401。
• require_editor 检查用户角色，身份明确但权限不足时使用 403。
• read_order 处理订单查询，不再重复安全细节。

这样拆开以后，公开接口可以不声明任何认证依赖，登录后可访问的接口使用 CurrentUserDep，编辑接口使用 EditorDep。同一套“识别当前用户”逻辑会被各层复用。

用可调用实例生成不同权限要求

角色一多，为每个角色复制一个 require_xxx 函数也会变得啰嗦。可以把“要求哪个角色”做成构造参数：

class RequireRole:
    def __init__(self, role: str):
        self.role = role
 
    def __call__(self, user: CurrentUserDep) -> User:
        if self.role not in user.roles:
            raise HTTPException(
                status_code=403,
                detail=f"需要{self.role}权限",
            )
        return user

FastAPI 要求依赖是“可调用对象”，函数、类和实现了 __call__ 的实例都符合条件。这里的实例在应用加载时保存角色配置；真正的 __call__ 仍然在请求期间执行，里面的 user 仍然由下层依赖注入。

安全依赖也会进入接口文档

OAuth2PasswordBearer 不只是读请求头。它还会在 OpenAPI 中登记安全方案，受保护接口会带上对应的安全要求，因此交互文档能显示授权入口。这里的 tokenUrl="token" 是告诉文档客户端去哪里获取令牌，并不会自动创建 /token 接口；令牌签发端点仍然需要你自己实现。

同步依赖和异步依赖可以混用

依赖函数和路由函数一样，既能写成 def，也能写成 async def。FastAPI 会识别两种形式：异步依赖在事件循环中等待，同步依赖由线程池执行，再把结果交回异步路由。

下面的同步依赖只读取请求头并拼出标签：

from typing import Annotated
 
from fastapi import Depends, Header
 
 
def request_label(
    x_request_id: Annotated[str, Header()] = "未提供",
) -> str:
    return f"请求-{x_request_id}"
 
 
@app.get("/mixed")
async def mixed(
    label: Annotated[str, Depends(request_label)],

请求：

GET /mixed
X-Request-ID: c42

返回：

{
  "label": "请求-c42",
  "handler": "异步路由"
}

同步依赖和异步路由并不冲突。选择写法时，看依赖内部调用的库：

• 库提供异步方法，需要用 await，依赖就写成 async def。
• 库只有阻塞式调用，依赖可以写成普通 def，让 FastAPI 在线程池中执行它。
• 单纯做少量内存计算，两种形式都能工作；保持代码直接、与所调用的 API 一致即可。

需要分清的是，线程池安排只发生在 FastAPI 主动调用的路由函数和依赖函数上。你在普通工具函数里直接调用另一个普通函数时，那仍然是一次普通 Python 调用；FastAPI 不会悄悄接管它。

用 yield 管好数据库会话

有些依赖不能只“返回一个值”就结束。数据库会话、文件、锁和临时客户端都需要收尾。理想流程是请求前获取资源，把资源交给接口，请求结束后无论成功还是异常都释放资源。

依赖函数把 return 换成一次 yield，就能把准备和清理写在一起：

from itertools import count
from typing import Annotated, Generator
 
from fastapi import Depends
 
 
events: list[str] = []
session_ids = count(1)
 
 
class DemoSession:
    def __init__(self, session_id: int):
        self.session_id = session_id
 
    def load_order(self, order_id: int

yield 前面的代码在接口执行前运行，yield 交出的 session 被注入接口，finally 里的关闭操作在使用结束后运行。一次成功的订单请求留下这样的事件顺序：

打开会话#1
查询订单#17
关闭会话#1

即使接口函数抛出异常，Python 仍然会进入 finally。这正是资源依赖最需要的保证：清理逻辑和创建逻辑放在同一个函数里，调用接口的人不必记住每条返回路径都要关闭会话。

把提交和回滚也放进边界

写操作通常还要决定事务成功还是失败。可以把事务边界放进依赖：

def get_session() -> Generator[Session, None, None]:
    session = Session(engine)
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

接口正常结束后执行 commit；接口或更上层依赖抛出异常时执行 rollback，随后用 raise 把原异常继续交给 FastAPI；不管前面发生什么，最后都关闭会话。

是否把 commit 自动放在依赖里，要跟项目的事务策略一致。一个请求可能包含多个业务动作时，统一提交很方便；如果某些接口需要更细的提交点，就应由服务层明确控制事务，依赖只负责提供和关闭会话。关键不是固定选一种写法，而是让事务边界清楚可见。

多层 yield 会按反方向清理

依赖可以继续依赖另一个带 yield 的依赖。假设服务对象依赖数据库会话，接口又依赖服务对象，创建顺序是“会话 → 服务 → 接口”，清理顺序会反过来变成“接口结束 → 服务清理 → 会话关闭”。这样上层资源在退出时仍能使用下层资源完成收尾。

清理发生在函数后还是响应后

带 yield 的依赖默认使用请求级作用域，退出代码在响应处理完成后运行。需要在接口函数返回后、响应发送前就释放资源时，可以明确写：

SessionDep = Annotated[
    DemoSession,
    Depends(get_session, scope="function"),
]

scope="function" 适合响应阶段不再需要该资源的情况。默认的 scope="request" 会把资源保持到请求与响应周期结束，流式响应还要在发送数据期间读取数据库时通常需要它。选择作用域时，要看响应生成阶段是否仍会使用资源，不能只为了“尽早关闭”就一律改成函数级。

use_cache 缓存的是同一次请求

嵌套依赖可能让同一个底层函数被多条路径需要。FastAPI 默认不会在一次请求里重复执行它，而是保存第一次得到的值，交给后续需要它的地方。

用一个每次调用都会递增的依赖，就能看清边界：

from itertools import count
from typing import Annotated
 
from fastapi import Depends
 
 
ticket_counter = count(1)
 
 
def issue_ticket() -> int:
    return next(ticket_counter)
 
 
TicketDep = Annotated[int, Depends(issue_ticket)]
FreshTicketDep = Annotated[
    int,
    Depends(issue_ticket, use_cache

第一次请求返回：

{
  "first": 1,
  "second": 1,
  "fresh": 2
}

first 触发第一次调用，得到 1；second 使用相同依赖，复用 1；fresh 明确设置 use_cache=False，于是再次调用并得到 2。

紧接着发送第二次请求：

{
  "first": 3,
  "second": 3,
  "fresh": 4
}

新请求没有沿用上一次请求缓存的 1。这里的缓存只服务于当前请求的依赖解析，不是进程级缓存，也不是跨用户缓存，更不是把依赖变成全局单例。

数据库会话和当前用户通常应该保留默认缓存。同一次请求里的多个服务拿到同一个会话、同一个用户，既减少重复工作，也保持事务视角一致。

use_cache=False 只在确实需要“每次走到这里都重新计算”时使用，例如读取会变化的瞬时值，或同一请求中明确要求两次独立采样。不要把它当成“关闭所有缓存”的性能开关；多数依赖的默认复用正是你想要的行为。

把依赖放在合适的层级

依赖可以声明在接口参数、路由装饰器、路由组和整个应用上。位置越高，影响范围越大。

接口参数：需要使用返回值

当前用户、分页对象、数据库会话都属于这一类：

@app.get("/orders/{order_id}")
async def read_order(
    order_id: int,
    editor: EditorDep,
    session: SessionDep,
):
    ...

签名直接展示了接口业务需要的输入，编辑器也能提供类型提示。

路由装饰器：只要求依赖成功

请求签名校验、内部密钥检查等工作常放在这里：

@app.post(
    "/hooks/payment",
    dependencies=[Depends(verify_signature)],
)
async def payment_hook(payload: PaymentEvent):
    ...

接口不使用校验函数的返回值，但校验失败会在业务处理前中止请求。

路由组：保护一批相关接口

from fastapi import APIRouter
 
 
admin_router = APIRouter(
    prefix="/admin",
    dependencies=[Depends(require_admin)],
)

挂到 admin_router 上的每个接口都会执行管理员检查。这样新增后台接口时，不容易忘记加保护。

整个应用：真正全局的前置要求

app = FastAPI(
    dependencies=[Depends(verify_gateway_header)],
)

全局依赖会作用于所有路由，包括以后新增的路由。健康检查、登录接口和公开接口是否也应该经过它，必须提前想清楚。范围越大，越要避免把只属于某个业务模块的规则提到全局。

一项依赖只做一件能命名的事

如果一个依赖同时解析令牌、查询用户、检查五种角色、打开数据库、写审计日志、调用外部服务，它虽然减少了接口函数里的代码，却把所有复杂度挤进了一个更难测试的函数。

更稳妥的拆法是：

• oauth2_scheme 提取令牌。
• get_current_user 识别用户。
• require_editor 判断权限。
• get_session 管理会话。
• write_audit_event 记录审计信息。

上层依赖可以组合这些小边界。每一层有清楚的输入、输出和失败方式，测试时也能只替换真正接触外部系统的那一层。

测试时替换依赖边界

依赖注入对测试最直接的帮助，不是“更容易创建模拟对象”这句抽象结论，而是 FastAPI 已经给了一个明确入口：app.dependency_overrides。

订单接口通常要验证令牌、查询用户、连接数据库。测试“有编辑权限时能读取订单”时，没有必要每次都走真实认证服务。可以用一个固定用户替换 get_current_user：

from fastapi.testclient import TestClient
 
from main import app, get_current_user, User
 
 
client = TestClient(app)
 
 
def test_editor_can_read_order():
    async def fake_current_user() -> User:
        return User(
            username="测试用户",
            roles=["reader", "editor"],
        )
 
    app.dependency_overrides[get_current_user] = fake_current_user
    try

覆盖字典的键必须是原依赖可调用对象，值是替代依赖。FastAPI 解析到 get_current_user 时，会改用 fake_current_user。原来的 get_current_user 以及它依赖的 oauth2_scheme 都不会执行，所以测试请求不必再提供 Authorization 请求头。

清除覆盖是测试的一部分

dependency_overrides 挂在应用对象上。如果测试结束后不清理，下一个测试可能继续拿到“测试用户”，结果会受到执行顺序影响。使用 try/finally、测试夹具，或在每个测试后统一 clear()，都比依赖人工记忆可靠。

清理后再次请求同一个受保护接口，但不带令牌，会恢复正常的认证结果：

{
  "detail": "Not authenticated"
}

状态码是 401。这也证明覆盖只在设置期间生效，没有修改原依赖函数。

覆盖边界，不要重写业务答案

好的替代依赖通常模拟系统边界：固定用户、测试数据库会话、假的邮件客户端、可控的时间提供器。接口函数和业务服务仍然按原路径运行，这样测试才会覆盖真正想验证的组合关系。

如果替代依赖直接返回接口最终 JSON，测试虽然很快，却绕开了权限判断、数据读取和响应拼装，能证明的东西很少。判断该覆盖哪一层时，可以问一句：我们是想隔离哪个昂贵、不稳定或不可控的外部边界？

这组示例覆盖了参数校验、嵌套认证、yield 清理、请求内缓存、同步与异步混用，以及依赖替换。测试结果为：

5 passed in 0.27s

从一条接口检查整张依赖图

写完一个带依赖的接口，不要只看最终能不能返回 200。顺着请求执行顺序检查一遍，更容易提前发现生命周期和权限问题。

几个常见误判

“设置了 use_cache=True 就会跨请求共享”是错的。默认依赖缓存只在一次请求的解析过程中复用。

“接口是 async def，所有依赖都必须是异步函数”也是错的。同步和异步可以混用，FastAPI 会分别安排；关键是依赖内部调用的库是否阻塞、是否需要 await。

“类型相同就会注入同一个对象”仍然是错的。Depends 里传入的可调用对象决定值从哪里来，类型标注负责说明这个值是什么。

“在接口末尾调用 session.close() 就够了”只覆盖了成功路径。接口提前抛错、权限依赖失败或响应生成异常时都可能绕过那一行。资源创建和清理应放在同一个 yield 依赖里。

收束到一个可维护的接口签名

回头看订单接口，它最后只留下三个业务上真正有意义的参数：订单编号、通过编辑权限检查的用户、已经打开的数据库会话。

@app.get("/orders/{order_id}")
async def read_order(
    order_id: int,
    editor: EditorDep,
    session: SessionDep,
):
    order = session.load_order(order_id)
    return {
        "operator": editor.username,
        "order": order,
        "session_id": session.session_id,
    }

这个签名本身就能读出接口的前置条件。FastAPI 负责沿图向下解析令牌、用户、权限和会话，遇到错误就提前返回；接口完成后，再沿相反方向清理资源。测试需要隔离外部系统时，替换对应依赖即可，业务函数不用改。

依赖注入在这里不是为了给简单代码套一层概念。它把原本散落在每个接口开头和结尾的工作，变成了有输入、有返回值、有失败方式、有生命周期的明确边界。下一次准备复制请求前置代码时，先问一句：这段工作能不能被命名成一个依赖？通常这就是拆分的起点。

FastAPI 调用 list_products(window=刚才的对象)，接口函数只处理筛选和分页，不再重复关心参数是怎样取得的。