首页
/ FastAPI 依赖注入进阶:使用 `yield` 的依赖(退出清理代码)完整解析

FastAPI 依赖注入进阶:使用 `yield` 的依赖(退出清理代码)完整解析

2026-09-06 18:21:59作者:廉彬冶Miranda

本文围绕 FastAPI 官方教程的"依赖与 yield"章节(见 docs/en/docs/tutorial/dependencies/dependencies-with-yield.md)展开,系统讲解如何在依赖函数中用 yield 代替 return,实现"响应结束后自动执行清理代码"的能力。读完本文,你将掌握:用 yield 依赖管理数据库会话的开关、在退出代码中感知与处理异常、子依赖树的正确退出顺序、Depends(scope="function") 提前退出的机制,以及上下文管理器(Context Manager)在依赖中的内部实现原理。

yield 依赖是 FastAPI 依赖注入体系中面向真实生产场景的核心能力,无论是"每个请求打开一个数据库会话并在结束时关闭",还是"用 finally 保证资源释放",都是通过本节语法完成的。下文将结合仓库中的源码示例(docs_src/dependencies/ 目录)与核心实现(fastapi/dependencies/utils.py)逐层讲解。

一、什么是"使用 yield 的依赖"

普通依赖在函数中通过 return 把值交给路径操作函数使用;而 FastAPI 支持一种更强大的依赖:依赖函数在 yield 之后还可以继续写代码,这部分代码会在"依赖的使用过程结束之后"再执行。它常被称作 exit code、cleanup code、teardown code、closing code 或 context manager exit code——名称很多,含义相同:收尾清理逻辑

async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

对这段代码的解读需要拆成三段来看(该文件即 docs_src/dependencies/tutorial007_py310.py):

  1. yield 之前的代码:在创建响应之前执行,例如 DBSession() 建立数据库会话;
  2. yield 产出的值:被注入到 路径操作函数 和其他依赖中,就像普通依赖的返回值一样;
  3. yield 之后的代码:在依赖的使用方(路径操作或下游依赖)执行完之后运行,例如 db.close() 关闭会话。

/// tip | 提示 每个依赖中请确保 yield 只出现一次。 ///

/// note | 技术细节 任何能用于 @contextlib.contextmanager@contextlib.asynccontextmanager 装饰的函数,都同样可以作为 FastAPI 依赖。事实上 FastAPI 内部正是借助这两个装饰器来完成转换(下文"源码印证"一节会给出证据)。 ///

源码印证:FastAPI 内部如何把依赖变成上下文管理器

仓库源码证实了上述注释的说法。在 fastapi/dependencies/utils.py_solve_generator 函数中:

async def _solve_generator(
    *, dependant: Dependant, stack: AsyncExitStack, sub_values: dict[str, Any]
) -> Any:
    assert dependant.call
    if _is_async_gen_callable(dependant.call):
        cm = asynccontextmanager(dependant.call)(**sub_values)
    elif _is_gen_callable(dependant.call):
        cm = contextmanager_in_threadpool(contextmanager(dependant.call)(**sub_values))
    return await stack.enter_async_context(cm)

(见 fastapi/dependencies/utils.py

  • 依赖函数是 async generator 时,FastAPI 用 asynccontextmanager(...) 包装;
  • 依赖函数是 普通(同步)generator 时,FastAPI 用 contextmanager(...) 包装,并通过 contextmanager_in_threadpool 放进线程池运行,避免阻塞事件循环;
  • 随后通过 AsyncExitStack.enter_async_context() 进入该上下文管理器,退出时即执行 yield 之后的代码。

因此你可以自由选择 async def 或普通 def 编写 yield 依赖,FastAPI 会像对待普通依赖一样自动处理两者的差异。

二、数据库会话依赖:yield 的经典用途

最常见的应用就是"创建数据库会话、用完后关闭"。上一节的 get_db 就是完整范式:

async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

结合本仓库的通用注入模式(docs_src/dependencies/tutorial007_py310.py),在路径操作函数中使用时通常写作:

from typing import Annotated
from fastapi import Depends

@app.get("/items/")
def read_items(db: Annotated[DBSession, Depends(get_db)]):
    ...
  • 请求到来时 FastAPI 先进入 get_db,执行到 yield db,把 db 注入给路径操作;
  • 路径操作执行期间 db 一直可用;
  • 路径操作返回后,FastAPI 退出该依赖,finally 中的 db.close() 保证会话被关闭——即使路径操作抛出了异常,finally 也一定会执行。

数据库会话需要同步与异步两种写法,FastAPI 均能正确处理:异步会话对象配 async def 依赖与 async with/await 关闭逻辑,同步会话配普通 def 依赖即可。

三、在 yield 依赖中使用 try 感知异常

yield 依赖中加入 try 块,可以接收"使用该依赖期间抛出的任何异常"。

例如:路径操作或其他依赖在运行中途触发了一次数据库事务回滚、或抛出了某个业务异常,你的 yield 依赖都能在 try 中捕获到它。于是可以针对特定异常写 except SomeException 分支做处理;同时用 finally 确保无论有无异常,退出步骤都照常执行。

上面 get_db 示例中的 try / finally 结构就是这种写法:finally 保证 db.close() 无论如何都会执行。这是最常用的资源清理范式,等价于在 with 语句退出时自动执行清理逻辑。

四、子依赖(sub-dependencies)与 yield:正确的退出顺序

你可以构建任意规模、任意形状的子依赖"树",且其中任意一个或全部依赖都能使用 yield。FastAPI 会保证每个 yield 依赖的"退出代码"按正确顺序运行。

仓库中的完整示例见 docs_src/dependencies/tutorial008_an_py310.py

from typing import Annotated
from fastapi import Depends


async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()


async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b: Annotated[DepB, Depends(dependency_b)]):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

依赖链为 dependency_c → dependency_b → dependency_a,三者都使用了 yield。执行顺序如下:

  • 进入阶段(正向)dependency_a 先启动到 yield,随后 dependency_b 依赖 dep_a 启动到 yield,最后 dependency_c 依赖 dep_b 启动到 yield
  • 退出阶段(反向)dependency_c 要执行退出代码,必须保证它引用的 dep_b(来自 dependency_b)仍然可用;同理 dependency_b 的退出代码需要 dep_a(来自 dependency_a)仍然可用。因此退出顺序必然是 dependency_c 先退出 → dependency_b 再退出 → dependency_a 最后退出

这就像把多个 with 语句嵌套起来:内层先创建、先使用,外层最后统一回收,形成正确的后进先出(LIFO)语义。

你还可以灵活组合:

  • 一部分依赖用 yield,另一部分用 return
  • 一个依赖同时依赖多个使用 yield 的子依赖;
  • 使用 yield 的依赖也可以被其他使用 yield 的依赖所依赖。

无论组合多复杂,FastAPI 都会保证一切按正确顺序执行。其机制正是 Python 的上下文管理器(Context Manager)——FastAPI 在内部利用 AsyncExitStack 把多个依赖的进入/退出编排成栈结构,天然保证"后进入的先退出"。

源码印证:依赖树求解与栈式退出

fastapi/dependencies/utils.pysolve_dependencies 中,每个子依赖在求值时如果是一个 generator callable(即使用了 yield),就会走 _solve_generatorenter_async_context 进入退出栈(见 fastapi/dependencies/utils.py):

elif _is_gen_callable(use_sub_dependant.call) or _is_async_gen_callable(
    use_sub_dependant.call
):
    use_astack = request_astack
    if sub_dependant.scope == "function":
        use_astack = function_astack
    solved = await _solve_generator(
        dependant=use_sub_dependant,
        stack=use_astack,
        sub_values=solved_result.values,
    )

AsyncExitStack 按"后进先出"顺序弹出并执行所有已注册的退出回调,这正是依赖树退出顺序正确的底层保证。

五、yield 依赖与 HTTPException

yield 依赖中同样可以使用 except 捕获被抛出的异常并做处理——例如把异常转换成 HTTPException。完整示例见 docs_src/dependencies/tutorial008b_an_py310.py

from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


data = {
    "plumbus": {"description": "Freshly pickled plumbus", "owner": "Morty"},
    "portal-gun": {"description": "Gun to create portals", "owner": "Rick"},
}


class OwnerError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except OwnerError as e:
        raise HTTPException(status_code=400, detail=f"Owner error: {e}")


@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):
    if item_id not in data:
        raise HTTPException(status_code=404, detail="Item not found")
    item = data[item_id]
    if item["owner"] != username:
        raise OwnerError(username)
    return item

执行流程:

  1. 路径操作函数 get_item 中发现物品所有者不是当前用户,抛出业务异常 OwnerError(username)
  2. 该异常会"反向传播"给 yield 依赖 get_username,触发其 except OwnerError as e 分支;
  3. 依赖把 OwnerError 重新包装为 HTTPException(status_code=400, detail=f"Owner error: {e}") 抛出;
  4. 客户端收到 400 Bad Request 及对应的错误信息。

/// tip | 提示 这属于比较进阶的技巧。多数场景下你并不需要它——完全可以在应用代码(例如路径操作函数内部)直接抛出 HTTPException。该能力主要是为"确实需要在依赖清理阶段统一转换异常"的场景准备的。 ///

如果希望捕获异常后生成自定义响应体,正确做法是注册一个 自定义异常处理器(Custom Exception Handler),让异常被集中转换。

六、except 吞掉异常的危险:必须重新 raise

yield 依赖中用 except 捕获异常后,如果不再抛出(既不重新抛出原异常、也不抛新异常),FastAPI 将无法感知到发生了异常——这与普通 Python 行为一致。示例见 docs_src/dependencies/tutorial008c_an_py310.py

class InternalError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except InternalError:
        print("Oops, we didn't raise again, Britney 😱")

后果:客户端虽然会收到应有的 HTTP 500 Internal Server Error 响应(因为没有抛出 HTTPException 之类的可识别异常),但服务器端没有任何日志、也没有任何错误痕迹——异常被彻底静默吞掉了,排查问题将非常困难。

黄金法则:except 中务必重新抛出

如果在 yield 依赖的 except 分支中不打算改抛 HTTPException 或其他新异常,必须用 raise 把原异常重新抛出。仓库示例见 docs_src/dependencies/tutorial008d_an_py310.py

class InternalError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except InternalError:
        print("We don't swallow the internal error here, we raise again 😎")
        raise


@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):
    if item_id == "portal-gun":
        raise InternalError(
            f"The portal gun is too dangerous to be owned by {username}"
        )
    ...

raise 会保留原始异常及其 traceback,客户端仍收到 HTTP 500 响应,而服务器日志中会完整记录自定义的 InternalError,方便监控与排障。

源码印证:为何"吞掉异常"会被特别关注

fastapi/routing.pyrequest_session 实现中,专门对"响应从未被 await"的情况做了检测,并抛出带有明确指引的 FastAPIError(见 fastapi/routing.py),错误信息明确指出:如果应用代码抛出了异常,而某个 yield 依赖里存在裸 exceptexcept Exception 且未再次抛出,就会触发该错误。这说明"吞异常"是 FastAPI 已知的高风险反模式,框架层面都为其准备了防御性检查。

七、yield 依赖的完整执行时序

理解"谁先执行、谁后执行"对正确使用 yield 依赖至关重要。原文档给出了如下时序图(时间自上而下,每一列是一个参与方):

sequenceDiagram

participant client as Client
participant handler as Exception handler
participant dep as Dep with yield
participant operation as Path Operation
participant tasks as Background tasks

    Note over client,operation: Can raise exceptions, including HTTPException
    client ->> dep: Start request
    Note over dep: Run code up to yield
    opt raise Exception
        dep -->> handler: Raise Exception
        handler -->> client: HTTP error response
    end
    dep ->> operation: Run dependency, e.g. DB session
    opt raise
        operation -->> dep: Raise Exception (e.g. HTTPException)
        opt handle
            dep -->> dep: Can catch exception, raise a new HTTPException, raise other exception
        end
        handler -->> client: HTTP error response
    end

    operation ->> client: Return response to client
    Note over client,operation: Response is already sent, can't change it anymore
    opt Tasks
        operation -->> tasks: Send background tasks
    end
    opt Raise other exception
        tasks -->> tasks: Handle exceptions in the background task code
    end

要点解读:

  1. 请求到达后,yield 依赖先运行 yield 之前的代码;
  2. 路径操作执行期间,任何一方(依赖或路径操作)都可以抛异常(包括 HTTPException);异常会传回 yield 依赖,由它决定捕获、改抛新异常还是继续抛出;
  3. 异常处理器最终把错误响应发给客户端;
  4. 一切正常时,路径操作把响应返回给客户端,yield 依赖随后执行退出代码;
  5. 后台任务(Background Tasks)在响应发出后由 Starlette 运行,它们的异常在任务自身代码内处理。

/// note | 关键约束 整个请求只会向客户端发送一个响应——要么是某个错误响应,要么是路径操作函数产生的正常响应。一旦其中任何一个响应已发出,就不能再发送第二个响应。这解释了为何 yield 依赖的退出代码运行在"响应之后"时,已无法再修改响应内容。 ///

/// tip | 实践建议 如果路径操作函数中抛出了任何异常(包括 HTTPException),该异常都会传回给各个 yield 依赖。大多数情况下,你应该在 yield 依赖中重新抛出同一个异常或抛出一个新异常,确保异常能被正确、完整地处理与记录。 ///

八、提前退出与 scope 参数

默认情况下,yield 依赖的退出代码是在响应发送给客户端之后执行的。但如果你确定"路径操作函数返回后就不再需要该依赖",可以使用 Depends(scope="function") 告诉 FastAPI:在路径操作函数返回后、响应发出之前就关闭该依赖。示例见 docs_src/dependencies/tutorial008e_an_py310.py

from typing import Annotated
from fastapi import Depends, FastAPI

app = FastAPI()


def get_username():
    try:
        yield "Rick"
    finally:
        print("Cleanup up before response is sent")


@app.get("/users/me")
def get_user_me(username: Annotated[str, Depends(get_username, scope="function")]):
    return username

Depends()scope 参数取值只有两种(对应 fastapi/params.py 中声明为 Literal["function", "request"] 的类型):

取值 开始时机 结束时机 语义
"function" 处理请求的路径操作函数开始前 路径操作函数结束后、响应发送回客户端之前 依赖的执行范围"包裹"着路径操作函数
"request" 处理请求的路径操作函数开始前(与 "function" 类似) 响应发送回客户端之后 依赖的执行范围"包裹"着整个请求与响应周期

默认值:当依赖使用了 yield 且未显式指定 scope 时,默认为 "request"

这一机制在你需要"确认写入提交成功后再真正返回给客户端"或"把清理工作提前到响应发出前完成"等场景非常有用。

scope 在子依赖中的约束

scope 参数对子依赖组合有以下规则:

  • 声明为 scope="request"(默认)的依赖,其所有子依赖也必须声明为 scope="request"
  • 声明为 scope="function" 的依赖,其子依赖可以是 scope="function",也可以是 scope="request"

原因在于:任何依赖都需要在自己的退出代码运行前,保证子依赖尚未退出——因为退出代码中可能还要使用子依赖提供的值。如果外层是"请求级"(要等响应发完才退出),而内层是"函数级"(路径操作一返回就退出),内层先退出后外层就再也用不到内层的值了,这与栈式生命周期矛盾。

仓库对此有硬性校验:在 fastapi/dependencies/utils.py,当"一个计算后 scoperequestyield 依赖"试图依赖一个 scope="function" 的子依赖时,会直接抛出 DependencyScopeError

raise DependencyScopeError(
    f'The dependency "{call_name}" has a scope of '
    '"request", it cannot depend on dependencies with scope "function".'
)

函数级与请求级退出栈的源码证据

两种 scope 在底层对应两套不同的 AsyncExitStack。在 fastapi/routing.pyrequest_session 中可以看到:外层 request_stack(请求级退出栈)包裹着内层 function_stack(函数级退出栈),内层函数栈在响应对象被 await(即真正发送响应)之前退出,而外层请求栈在响应发送之后才退出:

async with AsyncExitStack() as request_stack:
    scope["fastapi_inner_astack"] = request_stack
    async with AsyncExitStack() as function_stack:
        scope["fastapi_function_astack"] = function_stack
        response = await f(request)
    await response(scope, receive, send)

结合 fastapi/dependencies/utils.py 中从 request.scope 读取这两个栈的代码,可以看到:

  • scope="function"yield 依赖被 enterfunction_stack,随函数栈退出而提前关闭(在响应发出前);
  • scope="request"(默认)的 yield 依赖被 enterrequest_stack,响应发出后才关闭。

原文档给出了 scope="request"scope="function" 混用的时序图:

sequenceDiagram

participant client as Client
participant dep_req as Dep scope="request"
participant dep_func as Dep scope="function"
participant operation as Path Operation

    client ->> dep_req: Start request
    Note over dep_req: Run code up to yield
    dep_req ->> dep_func: Pass dependency
    Note over dep_func: Run code up to yield
    dep_func ->> operation: Run path operation with dependency
    operation ->> dep_func: Return from path operation
    Note over dep_func: Run code after yield
    Note over dep_func: Dependency closed
    dep_func ->> client: Send response to client
    Note over client: Response sent
    Note over dep_req: Run code after yield
    Note over dep_req: Dependency closed

九、yieldHTTPExceptionexcept 与后台任务的演进史

yield 依赖的能力是逐步演进的——它先后覆盖了不同使用场景,并修复了若干历史问题。官方把这段演进记录放在进阶指南的 Advanced Dependencies - Dependencies with yield, HTTPException, except and Background Tasks 一节。若你想了解某个版本的 FastAPI 中该特性如何变化、为何会有本文第五、六节描述的种种边界行为,建议阅读该章节。

十、深入底层:上下文管理器(Context Manager)

什么是上下文管理器

"上下文管理器"是指所有可以用于 with 语句的 Python 对象。以最常见的读文件为例:

with open("./somefile.txt") as f:
    contents = f.read()
    print(contents)

底层机制是:open("./somefile.txt") 返回的对象就是一个上下文管理器;当 with 代码块结束时——即使中途抛出了异常——它也会确保文件被正确关闭。

关键联系:当你编写一个 yield 依赖时,FastAPI 内部会为它创建一个上下文管理器,并将其与若干相关工具(例如依赖树求解、后台任务协调)组合使用。这就是"依赖 + yield"的底层本质:把函数依赖映射为上下文管理器生命周期

yield 依赖中使用自定义上下文管理器

/// warning | 进阶内容 这部分属于"进阶中的进阶"思路。如果你是 FastAPI 新手,可以先跳过本节,等熟练掌握基础用法后再回头阅读。 ///

在 Python 中,你可以通过实现 __enter__()__exit__() 两个方法创建一个上下文管理器类,并在 FastAPI 的 yield 依赖函数内部用 withasync with 使用它。仓库示例见 docs_src/dependencies/tutorial010_py310.py

class MySuperContextManager:
    def __init__(self):
        self.db = DBSession()

    def __enter__(self):
        return self.db

    def __exit__(self, exc_type, exc_value, traceback):
        self.db.close()


async def get_db():
    with MySuperContextManager() as db:
        yield db

这个例子的优雅之处在于:资源获取(DBSession())与释放(self.db.close())都被封装进上下文管理器自身;依赖函数只需 with MySuperContextManager() as db: 包裹住 yield db,即可把"会话对象"注入给路径操作,同时把"关闭会话"委托给上下文管理器,即便异常发生也能保证退出清理。

/// tip | 另一种创建上下文管理器的方式 还可以用装饰器方式创建上下文管理器:

用它们装饰一个只含单个 yield 的函数即可。这正是 FastAPI 内部对 yield 依赖所做的处理——但你不需要、也不应该在自己的 FastAPI 依赖上手动添加这些装饰器,FastAPI 会自动完成这一步(见本文第一节的 _solve_generator 源码)。 ///

十一、配套测试与继续深入

仓库为本教程每一节示例都配套了可运行的测试,集中在 tests/test_tutorial/test_dependencies/ 目录下:

  • test_tutorial007.py:覆盖数据库会话 yield 依赖的基础行为;
  • test_tutorial008.py:覆盖子依赖树 + yield 的退出顺序;
  • test_tutorial008b.py:覆盖 yield 依赖中捕获异常并改抛 HTTPException
  • test_tutorial008c.py / test_tutorial008d.py:覆盖"吞掉异常"与"重新 raise"两种行为差异;
  • test_tutorial008e.py:覆盖 Depends(scope="function") 的提前退出语义。

此外,仓库还有更细粒度的行为测试,例如 tests/test_dependency_after_yield_raise.pytests/test_dependency_yield_scope.pytests/test_dependency_contextmanager.py 等,分别验证异常传播、scope 生命周期与上下文管理器相关边界。阅读这些测试可以帮你快速建立"预期行为"的心智模型。

若想系统补足前置知识,可先阅读 依赖注入入门教程;若关心 yield 依赖在异常与后台任务叠加场景下的全部细节,请阅读 Advanced Dependencies

小结

yield 依赖把"依赖提供"与"资源清理"收进同一个函数,是 FastAPI 中编写数据库会话管理、事务边界、锁释放等代码的标准姿势。核心记忆点有四:

  1. yield 之前是前置逻辑,yield 产出的值注入路径操作,yield 之后是清理逻辑;
  2. 退出代码按依赖树的反向顺序执行,本质由 Python 上下文管理器与 AsyncExitStack 保证;
  3. except 中捕获异常后除非明确改抛新异常,否则必须 raise 重新抛出,避免错误被静默吞掉;
  4. 默认 scope="request" 使清理在响应发出后执行;若需提前清理可显式声明 Depends(..., scope="function"),但要遵循"request 级依赖不得依赖 function 级子依赖"的约束。
登录后查看全文
热门项目推荐
相关项目推荐