带 yield 的依赖

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

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

提示

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

技术细节

任何函数，如果可以与

• @contextlib.contextmanager 或
• @contextlib.asynccontextmanager

一起使用，就可以作为 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 块，尝试执行一些代码，然后在 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.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

如果你想捕获异常并基于此创建自定义响应，请创建自定义异常处理器。

带 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

信息

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

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

提示

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

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

带 yield 的依赖随着时间演变，以涵盖不同的用例并解决一些问题。

如果你想了解 FastAPI 不同版本中的变化，你可以在高级指南中阅读更多内容，在高级依赖 - 带 yield、HTTPException、except 和后台任务的依赖。

上下文管理器

什么是“上下文管理器”

“上下文管理器”是任何你可以用 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__() 的类来创建上下文管理器。

你也可以在带 yield 的 FastAPI 依赖中，通过在依赖函数内部使用 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

提示

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

• @contextlib.contextmanager 或
• @contextlib.asynccontextmanager

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

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

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

FastAPI 会在内部为你完成。