Dependencies with yield¶

FastAPI 支持带有可以在完成后执行一些额外步骤的依赖项。

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

提示

请确保每个依赖项最多使用一次 yield。

技术细节

任何适用于

的函数都可以用作 **FastAPI** 依赖项。

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

A database dependency with `yield`¶

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

只有在 yield 语句之前（包括该语句）的代码会在创建响应之前执行。

Python 3.9+

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

yield 的值将被注入到 *路径操作* 和其他依赖项中。

Python 3.9+

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

yield 语句之后的代码将在响应之后执行。

Python 3.9+

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

提示

您可以使用 async 函数或普通函数。

**FastAPI** 会正确处理这两者，与普通依赖项相同。

A dependency with `yield` and `try`¶

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

例如，如果在中间的某个点，在另一个依赖项或 *路径操作* 中，某个数据库事务被“回滚”或产生了任何其他异常，您将在您的依赖项中收到该异常。

因此，您可以在带有 except SomeException 的依赖项中查找该特定异常。

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

Python 3.9+

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

Sub-dependencies with `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.9+ - 非注解

提示

如果可能，请优先使用 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.9+ - 非注解

提示

如果可能，请优先使用 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 的 Context Managers。

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

Dependencies with `yield` and `HTTPException`¶

您已经看到，您可以将带有 yield 的依赖项与 try 块结合使用，这些块会尝试执行一些代码，然后在 finally 之后运行一些退出代码。

您还可以使用 except 来捕获引发的异常并对其进行处理。

例如，您可以引发另一个异常，例如 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.9+ - 非注解

提示

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

如果您想捕获异常并基于此创建自定义响应，请创建一个 Custom Exception Handler。

Dependencies with `yield` and `except`¶

如果您使用 except 在带有 yield 的依赖项中捕获异常，并且您没有再次引发它（或引发新异常），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.9+ - 非注解

提示

如果可能，请优先使用 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 Internal Server Error* 响应，正如预期的那样，因为我们没有引发 HTTPException 或类似的东西，但服务器 **不会有任何日志** 或其他错误指示。😱

Always `raise` in Dependencies with `yield` and `except`¶

如果您在带有 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.9+ - 非注解

提示

如果可能，请优先使用 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 Internal Server Error* 响应，但服务器将在日志中看到我们的自定义 InternalError。😎

Execution of dependencies with `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

信息

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

发送其中一个响应后，将无法再发送其他响应。

提示

如果您在 *路径操作函数* 的代码中引发任何异常，它将被传递给带有 yield 的依赖项，包括 HTTPException。在大多数情况下，您会希望从带有 yield 的依赖项中重新引发相同的异常或一个新的异常，以确保它得到妥善处理。

Early exit and `scope`¶

通常，带有 yield 的依赖项的退出代码会在响应发送给客户端 **之后** 执行。

但是，如果您知道在从 *路径操作函数* 返回后不再需要使用该依赖项，您可以使用 Depends(scope="function") 来告诉 FastAPI 它应该在 *路径操作函数* 返回后、**但在响应发送之前** 关闭该依赖项。

Python 3.9+

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

🤓 其他版本和变体

Python 3.9+ - 非注解

提示

如果可能，请优先使用 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: str = Depends(get_username, scope="function")):
    return username

Depends() 接收一个 scope 参数，它可以是

"function"：在处理请求的 *路径操作函数* 之前启动依赖项，在 *路径操作函数* 结束之后、**但在响应发送回客户端之前** 结束依赖项。因此，依赖项函数将在 *路径操作* **函数** *的周围* 执行。
"request"：在处理请求的 *路径操作函数* 之前启动依赖项（与使用 "function" 时类似），但在响应发送回客户端 **之后** 结束。因此，依赖项函数将在 **请求** 和响应周期 *的周围* 执行。

如果未指定并且依赖项带有 yield，则默认情况下其 scope 为 "request"。

`scope` for sub-dependencies

当您声明一个带有 scope="request"（默认值）的依赖项时，任何子依赖项也必须具有 scope 为 "request"。

但是，一个带有 scope 为 "function" 的依赖项可以有 scope 为 "function" 和 scope 为 "request" 的子依赖项。

这是因为任何依赖项都需要能够在子依赖项之前执行其退出代码，因为它可能需要在其退出代码中仍使用它们。

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

Dependencies with `yield`, `HTTPException`, `except` and Background Tasks¶

带有 yield 的依赖项已经随着时间的推移而发展，以覆盖不同的用例并修复一些问题。

如果您想了解 FastAPI 不同版本中的变化，可以在高级指南中阅读更多内容，在 Advanced Dependencies - Dependencies with yield, HTTPException, except and Background Tasks 中。

Context Managers

What are "Context Managers"

"Context Managers" 是任何可以在 with 语句中使用的 Python 对象。

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

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

底层，open("./somefile.txt") 创建了一个名为“Context Manager”的对象。

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

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

Using context managers in dependencies with `yield`¶

警告

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

如果您刚开始使用 **FastAPI**，您可能暂时可以跳过它。

在 Python 中，您可以通过创建具有两个方法的类：__enter__() 和 __exit__() 来创建上下文管理器。

您还可以通过在依赖项函数内部使用 with 或 async with 语句在 **FastAPI** 依赖项中使用它们。

Python 3.9+

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 会在内部为您处理。

Dependencies with yield¶

A database dependency with yield¶

A dependency with yield and try¶

Sub-dependencies with yield¶

Dependencies with yield and HTTPException¶

Dependencies with yield and except¶

Always raise in Dependencies with yield and except¶

Execution of dependencies with yield¶

Early exit and scope¶

scope for sub-dependencies

Dependencies with yield, HTTPException, except and Background Tasks¶

Context Managers

What are "Context Managers"

Using context managers in dependencies with yield¶

A database dependency with `yield`¶

A dependency with `yield` and `try`¶

Sub-dependencies with `yield`¶

Dependencies with `yield` and `HTTPException`¶

Dependencies with `yield` and `except`¶

Always `raise` in Dependencies with `yield` and `except`¶

Execution of dependencies with `yield`¶

Early exit and `scope`¶

`scope` for sub-dependencies

Dependencies with `yield`, `HTTPException`, `except` and Background Tasks¶

Using context managers in dependencies with `yield`¶