使用 yield 的依賴項¶
FastAPI 支援在完成後執行一些額外步驟的依賴項。
要做到這一點，請使用 yield 而不是 return，並在後面撰寫額外的步驟（程式碼）。
提示
確保每個依賴項只使用一次 yield。
「技術細節」
任何有效用於
都可以作為 FastAPI 的依賴項。
事實上，FastAPI 在內部使用了這兩個裝飾器。
使用 yield 的資料庫依賴項¶
例如，您可以使用它來建立資料庫工作階段並在完成後關閉它。
只有在 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 會在內部為您執行此操作。