首页
/ FastAPI 子依赖(Sub-dependencies)全解析:依赖的递归求解、每请求缓存与 use_cache 用法

FastAPI 子依赖(Sub-dependencies)全解析:依赖的递归求解、每请求缓存与 use_cache 用法

2026-09-07 13:12:03作者:鲍丁臣Ursa

导读

本文围绕 FastAPI 官方教程《Sub-dependencies》一篇展开(对应仓库中的 docs/hi/docs/tutorial/dependencies/sub-dependencies.md,其英文原版见 docs/en/docs/tutorial/dependencies/sub-dependencies.md),深入讲解“依赖(dependency)也可以拥有自己的依赖”这一机制。你将掌握如何让一个依赖函数同时充当“被依赖方(dependable)”与“依赖方(dependant)”、FastAPI 如何递归地求解任意深度的依赖图、为什么同一个子依赖在同一请求内只会被执行一次,以及如何通过 Depends(..., use_cache=False) 在高级场景中关闭这一缓存行为。文章在继承教程全部示例的基础上,结合当前仓库中 fastapi/dependencies/utils.pyfastapi/params.py 的真实源码,给出底层实现依据。

什么是子依赖(Sub-dependencies)

FastAPI 的依赖注入并不限制依赖只能有一层:你可以创建拥有自己 sub-dependencies 的依赖,而且嵌套深度完全由你决定——它们可以任意地 deep。开发者只需要声明依赖关系,剩下的求解工作全部交由 FastAPI 完成。

本教程用一个贯穿始终的完整示例(位于 docs_src/dependencies/tutorial005_an_py310.py)来演示这一机制,它包含三个角色:

  1. 第一层依赖 query_extractor:纯粹的“dependable”(被依赖方),从查询参数取数;
  2. 第二层依赖 query_or_cookie_extractor:既是一个 dependable,又是一个 dependant(它依赖前者);
  3. 路径操作函数 read_query:只声明了第二层依赖,却间接消费了两层的计算结果。

术语速记

为便于后文阅读,先明确教程中反复出现的两个概念:

  • dependable:一个可被其他函数依赖的组件(这里是普通的依赖函数);
  • dependant:一个本身声明了依赖、需要其他组件先求出结果的组件。

同一个函数可以同时是 dependable 和 dependant——这正是子依赖机制的本质。

第一层依赖:做一个纯粹的“dependable”

首先创建一个最简单的依赖函数 query_extractor,它声明一个可选的查询参数 q(类型为 str),然后原样返回它:

def query_extractor(q: str | None = None):
    return q

对应源码见 docs_src/dependencies/tutorial005_an_py310.py

这段代码“相当简单(并不是很有用)”,教程特意如此设计——正是为了让我们把注意力集中在子依赖的调用关系上,而不是业务逻辑上。它的唯一职责是:当 FastAPI 求解依赖图时,把请求中的 q 查询参数解析出来并作为返回值暴露给下一层。

第二层依赖:同时充当“dependable”与“dependant”

接下来定义第二个依赖函数 query_or_cookie_extractor。它自己声明了一个依赖(依赖第一层的 query_extractor),因此它既是 dependable 又是 dependant:

def query_or_cookie_extractor(
    q: Annotated[str, Depends(query_extractor)],
    last_query: Annotated[str | None, Cookie()] = None,
):
    if not q:
        return last_query
    return q

逐条分析它声明的参数:

  • 虽然这个函数本身是一个依赖(dependable),它同时也声明了另一个依赖(它 depends 于别的东西):
    • 它依赖 query_extractor,并把后者返回的值赋给参数 q
  • 它还声明了一个可选的 last_query cookie(类型为 str):
    • 如果用户本次请求没有提供查询参数 q,就回退使用上一次查询时写入 cookie 的值。

也就是说,第二层依赖实现了“查询参数优先,cookie 兜底”的取值策略。FastAPI 在调用它之前,会先自动求出 query_extractor 的值并注入到 q 形参中。

Annotated 写法的等价版本见 docs_src/dependencies/tutorial005_py310.py。教程建议:只要条件允许,优先使用 Annotated 版本。

完整的可运行示例

把两层依赖与路径操作组合起来,就是可以直接运行的最小应用(完整文件见 docs_src/dependencies/tutorial005_an_py310.py):

from typing import Annotated

from fastapi import Cookie, Depends, FastAPI

app = FastAPI()


def query_extractor(q: str | None = None):
    return q


def query_or_cookie_extractor(
    q: Annotated[str, Depends(query_extractor)],
    last_query: Annotated[str | None, Cookie()] = None,
):
    if not q:
        return last_query
    return q


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

运行方式与普通 FastAPI 应用一致,例如在安装好 fastapi 与 uvicorn 的环境中:

uvicorn docs_src.dependencies.tutorial005_an_py310:app

请求 GET /items/?q=hello 会返回 {"q_or_cookie":"hello"};不带 q 但携带了此前设置的 cookie 时,则会返回 cookie 中保存的上一次查询值。

在路径操作函数中使用依赖

最终在路径操作函数中,我们这样使用整个依赖链(见 docs_src/dependencies/tutorial005_an_py310.py):

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

注意一个关键事实:我们只向路径操作函数声明了一个依赖 query_or_cookie_extractor。但 FastAPI 会自行判断:要调用 query_or_cookie_extractor 并把结果传入,就必须先求解它的子依赖 query_extractor。整个求解链路可以用教程中的 Mermaid 图表示:

graph TB

query_extractor(["query_extractor"])
query_or_cookie_extractor(["query_or_cookie_extractor"])

read_query["/items/"]

query_extractor --> query_or_cookie_extractor --> read_query

即请求 /items/ 时,FastAPI 依次执行 query_extractorquery_or_cookie_extractorread_query,构成一条长度为 3 的调用链。

底层原理:依赖图是如何被递归求解的

教程所描述的“FastAPI 会搞定求解(solve)”,在当前仓库的依赖求解核心 fastapi/dependencies/utils.py 中有着清晰的代码对应。

每次处理请求时,FastAPI 都会调用 solve_dependencies,其核心是一个 for 循环遍历当前 dependant 的 dependencies 列表(即它的直接子依赖),见 fastapi/dependencies/utils.py

for sub_dependant in dependant.dependencies:
    ...
    solved_result = await solve_dependencies(
        request=request,
        dependant=use_sub_dependant,
        ...
        dependency_cache=dependency_cache,
        ...
    )

要点如下:

  1. 递归:对每一个 sub_dependantsolve_dependencies 都会再次调用它自己utils.py#L640),从而把“子依赖的子依赖……”层层展开——这正是“可以任意深度嵌套”的实现基础。
  2. 广度优先展开:循环先处理完当前层的所有子依赖,把结果写入一个 values 字典(通过 sub_dependant.name 作为键,见 utils.py#L677-L678),之后才去解析路径参数、查询参数、请求头、cookie、请求体等。
  3. 协程与线程池的适配:依赖如果是异步函数(_is_coroutine_callable),会被 await 调用;如果是普通同步函数,则通过 run_in_threadpool 放到线程池中执行,保证不会阻塞事件循环(见 utils.py#L673-L676)。

从源码结构可以看到,所谓“依赖求解”本质上是对一棵由 Dependant 节点构成的树做递归遍历:每个节点的 dependencies 列表就是它的子节点。教程里提到的“dependable / dependant”双重身份,在代码层面就是一个节点既可以出现在别人的 dependencies 里,自己也持有 dependencies

同一个依赖被多次声明:每请求只执行一次

教程接着指出一个重要的优化行为:

如果某个依赖对同一个 path operation 被多次声明——例如多个依赖共享同一个公共子依赖——FastAPI 会知道每个请求只需调用该子依赖一次,并把返回值存入一个“cache”,传给该请求内所有需要它的 dependants,而不是重复调用。

这一点同样能直接从源码得到印证。fastapi/dependencies/utils.py 中:

if sub_dependant.use_cache and sub_dependant_cache_key in dependency_cache:
    solved = dependency_cache[sub_dependant_cache_key]
elif _is_gen_callable(...) or _is_async_gen_callable(...):
    ...
elif _is_coroutine_callable(use_sub_dependant.call):
    solved = await call(**solved_result.values)
else:
    solved = await run_in_threadpool(call, **solved_result.values)
...
if sub_dependant_cache_key not in dependency_cache:
    dependency_cache[sub_dependant_cache_key] = solved

分析:

  • solve_dependencies 接收一个 dependency_cache 字典(默认值为空字典 {},见 utils.py#L615-L616),并在整个递归求解过程中把这个字典层层向下传递;
  • 每求解完一个子依赖,就依据依赖本身计算出一个缓存键 sub_dependant_cache_key,把结果写入 dependency_cache
  • 再次遇到相同的子依赖时,若 use_cacheTrue 且缓存键已命中,则直接复用缓存值,不再调用依赖函数。

需要特别说明的缓存作用域:从源码结构看,dependency_cache为单次请求的依赖求解而创建的局部字典,请求处理结束后即随该次调用消亡。因此这里的“缓存”是请求级缓存——它保证的是“同一个请求内不重复执行”,而不会把结果跨请求复用(跨请求复用本应是业务层的行为)。另外,凡是生成器形态的依赖(yield 形式,包括普通生成器与异步生成器),其生命周期由 async_exit_stack 管理(见 utils.py#L662-L672),这也是后置(teardown)代码能够按依赖层级依次执行的原因。

该默认参数定义在依赖声明类 Depends(以及派生自它的安全相关声明)的构造字段中,use_cache 的默认值为 True,见 fastapi/params.py

高级用法:use_cache=False 强制“每次现算”

在少数高级场景下,你可能确实需要同一个依赖在同一请求的每一步都被重新调用(甚至多次),而不是使用“缓存”中的值。此时可在使用 Depends 时把参数设为 use_cache=False

async def needy_dependency(fresh_value: Annotated[str, Depends(get_value, use_cache=False)]):
    return {"fresh_value": fresh_value}

Annotated 写法等价版本:

async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)):
    return {"fresh_value": fresh_value}

教程提示:如无特殊原因,优先使用 Annotated 写法。

对照前面的源码逻辑,设置 use_cache=False 后,缓存命中判断分支 if sub_dependant.use_cache and sub_dependant_cache_key in dependency_cache 将不会走捷径,依赖函数会在此次求解中被重新执行,从而拿到“新鲜”的值。

需要理解的是,use_cache 控制的是读取侧是否命中缓存,而不是彻底关闭记录。因此设计依赖时应保持依赖函数本身的幂等性/纯度:在默认配置下,一个请求内同一个依赖只会被执行一次,后续消费方共享首次执行的结果。

小结与后续阅读指引

抛开所有花哨的术语,依赖注入(Dependency Injection)系统其实相当简单

  • 它只是一些“长得和路径操作函数一样”的普通函数;
  • 但它非常强大,允许你声明任意深度嵌套的依赖“图”(树);
  • 求解顺序、结果注入、请求级缓存都由 FastAPI 内部统一处理,开发者只需声明“谁依赖谁”。

这些简单示例也许还看不出子依赖的威力,但在本仓库教程体系中紧接其后的 security(安全)相关章节里,你会发现这一机制的巨大价值——OAuth2、API Key 等各类安全组件都以“可复用的依赖”形式存在,子依赖让组合式鉴权逻辑得以优雅落地,并为你省下大量样板代码。

扩展阅读(仓库内可继续深入的文件)

登录后查看全文
热门项目推荐
相关项目推荐