yield を持つ依存関係¶
FastAPI は、終了後にいくつかの追加ステップを実行する依存関係をサポートしています。
これを行うには、return の代わりに yield を使用し、追加のステップ（コード）をその後に記述します。
ヒント
依存関係ごとに yield を一度だけ使用するようにしてください。
技術的な詳細
以下のいずれかで使用できるすべての関数
は、FastAPI の依存関係として使用できます。
実際、FastAPI は内部的にこれらの 2 つのデコレータを使用しています。
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()
ヒント
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 を持つ依存関係を使用し、いくつかのコードを実行しようとしてから finally の後にいくつかの終了コードを実行する try ブロックを持つことができることを見ました。
また、except を使用してスローされた例外をキャッチし、それを使って何かを行うこともできます。
たとえば、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
例外をキャッチし、それに基づいてカスタム応答を作成したい場合は、カスタム例外ハンドラーを作成してください。
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
この場合、HTTPException などが再発生されていないため、クライアントは期待通りHTTP 500 Internal Server Error応答を受け取りますが、サーバーはエラーのログやその他の兆候を一切持ちません。😱
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 Internal Server Error 応答を受け取りますが、サーバーはログにカスタムの 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 を持つ依存関係に渡されます。ほとんどの場合、適切に処理されることを確実にするために、yield を持つ依存関係から同じ例外または新しい例外を再発生させたいでしょう。
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 を持つ依存関係でコンテキストマネージャーを使用する¶
Warning
これは、多かれ少なかれ「高度な」アイデアです。
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 が内部的に処理します。