带有 `yield` 的依赖¶

FastAPI 支持在完成后执行一些额外步骤的依赖。

为此，请使用 `yield` 而非 `return`，并在其后编写额外的步骤（代码）。

提示

请确保每个依赖只使用一次 `yield`。

技术细节

任何有效函数，只要可以用于

都可以用作 FastAPI 依赖。

事实上，FastAPI 内部使用了这两个装饰器。

带有 `yield` 的数据库依赖¶

例如，你可以用它来创建数据库会话并在完成后关闭它。

只有 `yield` 语句之前和包含 `yield` 语句的代码会在创建响应之前执行

Python 3.8+

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

`yield` 的值会被注入到 *路径操作* 和其他依赖中

Python 3.8+

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

`yield` 语句后面的代码会在创建响应后但在发送响应前执行

Python 3.8+

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

提示

你可以使用 `async` 或普通函数。

FastAPI 将对它们进行正确处理，与处理普通依赖的方式相同。

带有 `yield` 和 `try` 的依赖¶

如果在带有 `yield` 的依赖中使用 `try` 块，你将收到在使用该依赖时抛出的任何异常。

例如，如果在中间某个点，在另一个依赖或 *路径操作* 中，某个代码导致数据库事务“回滚”或产生任何其他错误，你将在你的依赖中收到该异常。

因此，你可以在依赖内部使用 `except SomeException` 来查找该特定异常。

同样，你可以使用 `finally` 来确保退出步骤被执行，无论是否发生异常。

Python 3.8+

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

带有 `yield` 的子依赖¶

你可以拥有任意大小和形状的子依赖和子依赖“树”，其中任何或所有都可以使用 `yield`。

FastAPI 将确保带有 `yield` 的每个依赖中的“退出代码”以正确的顺序运行。

例如，`dependency_c` 可以依赖于 `dependency_b`，而 `dependency_b` 依赖于 `dependency_a`

Python 3.9+

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)

🤓 其他版本和变体

Python 3.8+Python 3.8+ - 非 Annotated

from fastapi import Depends
from typing_extensions import Annotated


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)

提示

如果可能，请优先使用 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=Depends(dependency_a)):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b=Depends(dependency_b)):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

并且它们都可以使用 `yield`。

在这种情况下，`dependency_c` 要执行其退出代码，需要 `dependency_b`（这里命名为 `dep_b`）的值仍然可用。

反过来，`dependency_b` 需要 `dependency_a`（这里命名为 `dep_a`）的值对其退出代码可用。

Python 3.9+

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)

🤓 其他版本和变体

Python 3.8+Python 3.8+ - 非 Annotated

from fastapi import Depends
from typing_extensions import Annotated


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)

提示

如果可能，请优先使用 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=Depends(dependency_a)):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b=Depends(dependency_b)):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

同样，你可以拥有一些带有 `yield` 的依赖和一些带有 `return` 的依赖，并且让其中一些依赖于另一些。

你还可以有一个依赖需要多个其他带有 `yield` 的依赖，等等。

你可以拥有任何你想要的依赖组合。

FastAPI 将确保一切都以正确的顺序运行。

技术细节

这得益于 Python 的上下文管理器。

FastAPI 内部使用它们来实现这一点。

带有 `yield` 和 `HTTPException` 的依赖¶

你已经看到可以使用带有 `yield` 的依赖，并拥有捕获异常的 `try` 块。

同样，你可以在退出代码中，`yield` 之后抛出 `HTTPException` 或类似异常。

提示

这是一种比较高级的技术，在大多数情况下你并不真的需要它，因为你可以从应用程序代码的其他部分（例如，在 *路径操作函数* 中）抛出异常（包括 `HTTPException`）。

但如果你需要，它就在那里。🤓

Python 3.9+

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

🤓 其他版本和变体

Python 3.8+Python 3.8+ - 非 Annotated

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

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

提示

如果可能，请优先使用 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: 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

另一种捕获异常（并可能也抛出另一个 `HTTPException`）的方法是创建自定义异常处理器。

带有 `yield` 和 `except` 的依赖¶

如果你在带有 `yield` 的依赖中使用 `except` 捕获异常，并且没有再次抛出它（或抛出新异常），FastAPI 将无法感知到异常的发生，这与普通 Python 的行为相同。

Python 3.9+

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


class InternalError(Exception):
    pass


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


@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}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id

🤓 其他版本和变体

Python 3.8+Python 3.8+ - 非 Annotated

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

app = FastAPI()


class InternalError(Exception):
    pass


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


@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}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id

提示

如果可能，请优先使用 Annotated 版本。

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


class InternalError(Exception):
    pass


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


@app.get("/items/{item_id}")
def get_item(item_id: str, username: str = Depends(get_username)):
    if item_id == "portal-gun":
        raise InternalError(
            f"The portal gun is too dangerous to be owned by {username}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id

在这种情况下，客户端将按预期收到 *HTTP 500 内部服务器错误* 响应，因为我们没有抛出 `HTTPException` 或类似异常，但服务器将 没有任何日志 或其他指示错误的信息。😱

总是在带有 `yield` 和 `except` 的依赖中 `raise` 异常¶

如果你在带有 `yield` 的依赖中捕获了异常，除非你抛出另一个 `HTTPException` 或类似异常，否则你应该重新抛出原始异常。

你可以使用 `raise` 重新抛出相同的异常

Python 3.9+

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


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}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id

🤓 其他版本和变体

Python 3.8+Python 3.8+ - 非 Annotated

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

app = FastAPI()


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}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id

提示

如果可能，请优先使用 Annotated 版本。

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


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: str = Depends(get_username)):
    if item_id == "portal-gun":
        raise InternalError(
            f"The portal gun is too dangerous to be owned by {username}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id

现在客户端将收到相同的 *HTTP 500 内部服务器错误* 响应，但服务器日志中将包含我们的自定义 `InternalError`。😎

带有 `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

信息

只会向客户端发送 一个响应。它可能是错误响应之一，也可能是来自 *路径操作* 的响应。

一旦发送了其中一个响应，就不能再发送其他响应。

提示

此图显示了 `HTTPException`，但你也可以抛出在带有 `yield` 的依赖中或通过自定义异常处理器捕获的任何其他异常。

如果你抛出任何异常，它都将被传递给带有 yield 的依赖，包括 `HTTPException`。在大多数情况下，你会希望重新抛出相同的异常或从带有 `yield` 的依赖中抛出新异常，以确保其得到正确处理。

带有 `yield`、`HTTPException`、`except` 和后台任务的依赖¶

警告

你很可能不需要这些技术细节，可以跳过本节并继续阅读下文。

这些细节主要在你使用 0.106.0 之前的 FastAPI 版本并在后台任务中使用了带有 `yield` 的依赖中的资源时有用。

带有 `yield` 和 `except` 的依赖，技术细节¶

在 FastAPI 0.110.0 之前，如果你使用带有 `yield` 的依赖，然后在该依赖中使用 `except` 捕获异常，并且没有再次抛出异常，那么该异常将自动抛出/转发给任何异常处理器或内部服务器错误处理器。

此行为在 0.110.0 版本中已更改，以修复由于没有处理器的转发异常（内部服务器错误）导致的未处理内存消耗，并使其与普通 Python 代码的行为保持一致。

后台任务和带有 `yield` 的依赖，技术细节¶

在 FastAPI 0.106.0 之前，在 `yield` 之后抛出异常是不可能的，因为带有 `yield` 的依赖中的退出代码是在响应发送 *之后* 执行的，所以异常处理器已经运行。

这样设计主要是为了允许在后台任务内部使用依赖“yield”的相同对象，因为退出代码会在后台任务完成后执行。

然而，由于这意味着在响应通过网络传输时，不必要地持有带有 `yield` 的依赖中的资源（例如数据库连接），因此在 FastAPI 0.106.0 中更改了此行为。

提示

此外，后台任务通常是一组独立的逻辑，应该单独处理，拥有自己的资源（例如，自己的数据库连接）。

因此，这样你可能会有更简洁的代码。

如果你曾经依赖于此行为，现在你应该在后台任务本身内部创建后台任务所需的资源，并且内部只使用不依赖于带有 `yield` 的依赖的资源的数据。

例如，你不会使用相同的数据库会话，而是在后台任务内部创建一个新的数据库会话，然后使用这个新会话从数据库获取对象。然后，不是将数据库中的对象作为参数传递给后台任务函数，而是传递该对象的 ID，然后在后台任务函数内部再次获取该对象。

上下文管理器¶

什么是“上下文管理器”¶

“上下文管理器”是任何可以在 `with` 语句中使用的 Python 对象。

例如，你可以使用 `with` 读取文件

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

在底层，`open("./somefile.txt")` 创建了一个被称为“上下文管理器”的对象。

当 `with` 块结束时，它会确保关闭文件，即使发生了异常。

当你创建一个带有 `yield` 的依赖时，FastAPI 将在内部为其创建一个上下文管理器，并将其与一些其他相关工具结合使用。

在带有 `yield` 的依赖中使用上下文管理器¶

警告

这或多或少是一个“高级”概念。

如果你刚开始使用 FastAPI，你可能现在想跳过它。

在 Python 中，你可以通过创建一个包含两个方法：`__enter__()` 和 `__exit__()` 的类来创建上下文管理器。

你也可以在 FastAPI 带有 `yield` 的依赖中，通过在依赖函数内部使用 `with` 或 `async with` 语句来使用它们。

Python 3.8+

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

提示

创建上下文管理器的另一种方法是使用

用单个 `yield` 装饰函数。

这就是 FastAPI 内部用于带有 `yield` 的依赖的方式。

但是你不必为 FastAPI 依赖使用装饰器（也不应该使用）。

FastAPI 会在内部为你完成。