yield を使用する依存関係¶
FastAPI は、終了後に追加のステップを実行する 依存関係をサポートしています。
これを行うには、`return` の代わりに `yield` を使用し、追加のステップ（コード）をその後に記述します。
ヒント
依存関係ごとに `yield` を 1 回だけ使用してください。
「技術的な詳細」
次のものと組み合わせて使用できる有効な関数
は、**FastAPI** 依存関係として使用できます。
実際、FastAPI は内部でこれら 2 つのデコレータを使用しています。
`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()
ヒント
`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 を使用した依存関係の実行¶
実行シーケンスは、この図とほぼ同じです。時間は上から下に流れます。各列は、コードを操作または実行する部分の1つです。
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情報
クライアントには**1つのレスポンス**のみが送信されます。エラーレスポンスの1つであるか、*パス操作*からのレスポンスになります。
これらのレスポンスの1つが送信された後、他のレスポンスを送信することはできません。
ヒント
この図は HTTPException を示していますが、yield を使用した依存関係または カスタム例外ハンドラ でキャッチした他の例外を発生させることもできます。
例外が発生した場合、HTTPException を含め、yield を使用した依存関係に渡されます。ほとんどの場合、適切に処理されるように、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 を使用してファイルを読み取ることができます
with open("./somefile.txt") as f:
contents = f.read()
print(contents)
内部的には、open("./somefile.txt") は「コンテキストマネージャー」と呼ばれるオブジェクトを作成します。
with ブロックが終了すると、例外が発生した場合でも、ファイルが確実に閉じられます。
yield を使用して依存関係を作成すると、**FastAPI** は内部的にコンテキストマネージャーを作成し、他の関連ツールと組み合わせます。
yield を使用した依存関係でのコンテキストマネージャーの使用¶
警告
これは、多かれ少なかれ「高度な」アイデアです。
**FastAPI** を使い始めたばかりの場合は、今はスキップすることをお勧めします。
Python では、__enter__() と __exit__() の2つのメソッドを持つクラスを作成することで、コンテキストマネージャーを作成できます。
依存関係関数内で with または async with ステートメントを使用して、**FastAPI** 依存関係内で yield と共に使用することもできます。
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 は内部的にそれを実行します。