带有 yield 的依赖项¶

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

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

Tip

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

"技术细节"

任何可以与以下内容一起使用的函数：

@contextlib.contextmanager 或
@contextlib.asynccontextmanager

都可以作为 FastAPI 依赖项使用。

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

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

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

只有在创建响应之前执行的代码是 yield 语句之前的代码：

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

yield 的值是被注入到路径操作和其他依赖项中的值：

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

yield 语句之后的代码在响应发送后执行：

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

Tip

你可以使用 async 或常规函数。

FastAPI 会根据情况正确处理它们，就像处理普通依赖项一样。

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

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

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

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

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

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+Python 3.8+Python 3.8+ non-Annotated

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)

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)

Tip

如果可能，建议使用 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+Python 3.8+Python 3.8+ non-Annotated

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)

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)

Tip

如果可能，建议使用 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 或类似的异常。

Tip

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

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

Python 3.9+Python 3.8+Python 3.8+ non-Annotated

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

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

Tip

如果可能，建议使用 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+Python 3.8+Python 3.8+ non-Annotated

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

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

Tip

如果可能，建议使用 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+Python 3.8+Python 3.8+ non-Annotated

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

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

Tip

如果可能，建议使用 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: 可以引发异常，包括 HTTPException
    client ->> dep: 开始请求
    Note over dep: 运行代码直到 yield
    opt 引发异常
        dep -->> handler: 引发异常
        handler -->> client: HTTP 错误响应
    end
    dep ->> operation: 运行依赖项，例如 DB 会话
    opt 引发
        operation -->> dep: 引发异常（例如 HTTPException）
        opt 处理
            dep -->> dep: 可以捕获异常，引发新的 HTTPException，引发其他异常
        end
        handler -->> client: HTTP 错误响应
    end

    operation ->> client: 返回响应给客户端
    Note over client,operation: 响应已经发送，不能再更改
    opt 任务
        operation -->> tasks: 发送后台任务
    end
    opt 引发其他异常
        tasks -->> tasks: 处理后台任务代码中的异常
    end

Info

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

在发送其中一个响应后，不能再发送其他响应。

Tip

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

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

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

Warning

你很可能不需要这些技术细节，可以跳过这一部分，继续往下阅读。

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

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

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

这一行为在 0.110.0 版本中进行了更改，以修复在没有处理器的情况下转发异常导致的未处理的内存消耗（内部服务器错误），并使其与常规 Python 代码的行为保持一致。

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

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

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

尽管如此，由于这意味着在依赖项中不必要地持有资源（例如数据库连接）时等待响应通过网络传输，这一行为在 FastAPI 0.106.0 中进行了更改。

Tip

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

因此，这种方式你可能会得到更清晰的代码。

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

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

上下文管理器¶

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

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

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

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

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

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

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

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

Warning

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

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

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

你也可以在 FastAPI 的依赖项中使用 yield，通过在依赖项函数内部使用 with 或 async with 语句：

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

Tip

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

@contextlib.contextmanager 或
@contextlib.asynccontextmanager

使用它们来装饰一个带有单个 yield 的函数。

这就是 FastAPI 在内部为带有 yield 的依赖项所做的事情。

但你不需要为 FastAPI 依赖项使用这些装饰器（也不应该使用）。

FastAPI 会在内部为你完成这些工作。

带有 yield 的依赖项¶

带有 yield 的数据库依赖项¶

带有 yield 和 try 的依赖项¶

带有 yield 的子依赖项¶

带有 yield 和 HTTPException 的依赖项¶

带有 yield 和 except 的依赖项¶

在带有 yield 和 except 的依赖项中始终 raise¶

带有 yield 的依赖项的执行顺序¶

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

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

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

上下文管理器¶

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

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

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

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

带有 `yield` 的子依赖项¶

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

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

在带有 `yield` 和 `except` 的依赖项中始终 `raise`¶

带有 `yield` 的依赖项的执行顺序¶

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

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

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

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