带 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 块来尝试执行一些代码，然后在 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 的依赖中重新抛出相同的异常或一个新的异常，以确保它得到妥善处理。

提前退出和 `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.8+Python 3.8+ - 非 Annotated

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

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

提示

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

当你声明一个 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

带 `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__() 的类来创建上下文管理器。

你也可以在 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 会在内部为你完成。

带 yield 的依赖¶

带 yield 的数据库依赖¶

带 yield 和 try 的依赖¶

带 yield 的子依赖¶

带 yield 和 HTTPException 的依赖¶

带 yield 和 except 的依赖¶

在带 yield 和 except 的依赖中总是要 raise¶

带 yield 的依赖的执行过程¶

提前退出和 scope¶

子依赖的 scope¶

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

上下文管理器¶

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

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

带 `yield` 的数据库依赖¶

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

带 `yield` 的子依赖¶

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

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

在带 `yield` 和 `except` 的依赖中总是要 `raise`¶

带 `yield` 的依赖的执行过程¶

提前退出和 `scope`¶

子依赖的 `scope`¶

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

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