使用 yield 的依赖项¶
FastAPI 支持在完成后执行一些额外步骤的依赖项。
为此，请使用 yield 代替 return，并在其后编写额外步骤（代码）。
Tip
确保每个依赖项只使用一次 yield。
技术细节
任何可以有效地与以下内容一起使用的函数：
都可以作为 FastAPI 依赖项有效使用。
实际上，FastAPI 在内部使用了这两个装饰器。
使用 yield 的数据库依赖项¶
例如，您可以使用它来创建数据库会话并在完成后关闭它。
只有 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：
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)
🤓 Other versions and variants
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
Prefer to use the Annotated version if possible.
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）的值来执行其退出代码。
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)
🤓 Other versions and variants
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
Prefer to use the Annotated version if possible.
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 将确保一切以正确的顺序运行。
使用 yield 和 HTTPException 的依赖项¶
您已经看到可以使用带有 yield 的依赖项并拥有捕获异常的 try 块。
同样，您可以在退出代码中，在 yield 之后引发 HTTPException 或类似异常。
Tip
这是一种有些高级的技术，在大多数情况下您并不真正需要它，因为您可以从应用程序代码的其余部分内部引发异常（包括 HTTPException），例如，在路径操作函数中。
但如果您需要，它就在那里。🤓
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
🤓 Other versions and variants
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
Prefer to use the Annotated version if possible.
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 中的情况相同：
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
🤓 Other versions and variants
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
Prefer to use the Annotated version if possible.
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 重新引发相同的异常：
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
🤓 Other versions and variants
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
Prefer to use the Annotated version if possible.
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
endInfo
只会向客户端发送一个响应。它可能是错误响应之一，也可能是来自路径操作的响应。
在发送了其中一个响应之后，不能再发送其他响应。
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
创建上下文管理器的另一种方法是使用：
使用它们来装饰带有单个 yield 的函数。
这就是 FastAPI 在内部用于带有 yield 的依赖项的方式。
但您不必为 FastAPI 依赖项使用装饰器（也不应该）。
FastAPI 会在内部为您处理。