带有 yield 的依赖项¶
FastAPI 支持在完成某些额外的步骤后的依赖项。
为此，使用 yield 而不是 return，并在之后编写额外的步骤（代码）。
提示
确保每个依赖项仅使用 yield 一次。
"技术细节"
任何与
一起使用有效的函数都将有效地用作 FastAPI 依赖项。
实际上，FastAPI 在内部使用这两个装饰器。
带有 yield 的数据库依赖项¶
例如，你可以使用它来创建数据库会话并在完成之后关闭它。
只有在创建响应之前执行的代码（包括 yield 语句）
async def get_db():
db = DBSession()
try:
yield db
finally:
db.close()
产生的值是注入路径操作和其他依赖项的值
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()
提示
你可以使用 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)
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）的值可用，才能执行其退出代码。
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)
提示
如果可能，请优先使用 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 会确保一切以正确的顺序运行。
带有 yield 和 HTTPException 的依赖项¶
您已经看到可以使用带有 yield 的依赖项并拥有捕获异常的 try 块。
同样，您可以在 yield 之后，在退出代码中抛出 HTTPException 或类似的异常。
提示
这是一种比较高级的技术，在大多数情况下您并不真正需要它，因为您可以在应用程序代码的其他部分中抛出异常（包括 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
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
您可以用来捕获异常（并可能还会抛出另一个 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
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 重新抛出相同的异常
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
提示
如果可能，请优先使用 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信息
只会将一个响应发送给客户端。它可能是错误响应之一，也可能是来自路径操作的响应。
在这些响应之一被发送之后，就无法再发送其他响应了。
提示
此图表显示了 HTTPException，但您也可以抛出您在带有 yield 的依赖项或 自定义异常处理程序 中捕获的任何其他异常。
如果您抛出任何异常，它将被传递给带有 yield 的依赖项，包括 HTTPException。在大多数情况下，您需要重新抛出同一个异常或从带有 yield 的依赖项中抛出另一个新异常，以确保它被正确处理。
带有 yield、HTTPException、except 和后台任务的依赖项¶
警告
您很可能不需要这些技术细节，您可以跳过本节，继续阅读下面的内容。
这些细节主要在您使用的是 0.106.0 之前的 FastAPI 版本，并且在后台任务中使用了来自带有 yield 的依赖项的资源时才有用。
带有 yield 和 except 的依赖项，技术细节¶
在 FastAPI 0.110.0 之前，如果您使用的是带有 yield 的依赖项，然后在该依赖项中使用 except 捕获异常，并且没有再次抛出异常，该异常将被自动抛出/转发给任何异常处理程序或内部服务器错误处理程序。
这在 0.110.0 版本中进行了更改，以修复未处理的内存消耗，这些消耗来自于未被处理程序处理的转发异常（内部服务器错误），并使其行为与普通 Python 代码保持一致。
后台任务和带有 yield 的依赖项，技术细节¶
在 FastAPI 0.106.0 之前，在 yield 之后抛出异常是不可能的，带有 yield 的依赖项中的退出代码是在发送响应之后执行的，因此 异常处理程序 已经运行过了。
这样设计的主要目的是允许在后台任务中使用依赖项 "yield" 的相同对象，因为退出代码将在后台任务完成后执行。
然而，由于这将意味着在不必要地持有带有 yield 的依赖项中的资源（例如数据库连接）的情况下等待响应通过网络传输，所以在 FastAPI 0.106.0 中进行了更改。
提示
此外，后台任务通常是一组独立的逻辑，应该单独处理，并有自己的资源（例如，自己的数据库连接）。
因此，这样您将获得更简洁的代码。
如果您曾经依赖这种行为，现在您应该在后台任务本身内部创建后台任务的资源，并在内部只使用不依赖于带有 yield 的依赖项资源的数据。
例如，您不再使用同一个数据库会话，而是在后台任务内部创建一个新的数据库会话，并使用这个新的会话从数据库中获取对象。然后，您不再将来自数据库的对象作为参数传递给后台任务函数，而是传递该对象的 ID，然后在后台任务函数内部再次获取该对象。
上下文管理器¶
"上下文管理器" 是什么¶
"上下文管理器" 是您可以用在 with 语句中的任何 Python 对象。
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 语句。
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 会在内部为您完成这一切。