コンテンツにスキップ

yieldを持つ依存関係

🌐 Translation by AI and humans

This translation was made by AI guided by humans. 🤝

It could have mistakes of misunderstanding the original meaning, or looking unnatural, etc. 🤖

You can improve this translation by helping us guide the AI LLM better.

English version

FastAPIは、いくつかの終了後の追加のステップを行う依存関係をサポートしています。

これを行うには、returnの代わりにyieldを使い、その後に追加のステップ（コード）を書きます。

豆知識

yieldは必ず依存関係ごとに1回だけ使用するようにしてください。

技術詳細

以下と一緒に使用できる関数なら何でも有効です:

これらは FastAPI の依存関係として使用するのに有効です。

実際、FastAPIは内部的にこれら２つのデコレータを使用しています。

yieldを持つデータベースの依存関係

例えば、これを使ってデータベースセッションを作成し、終了後にそれを閉じることができます。

レスポンスを作成する前に、yield文より前のコード（およびyield文を含む）が実行されます:

async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

生成された値は、path operationsや他の依存関係に注入されるものです:

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ブロックを使用した場合、その依存関係を使用した際にスローされたあらゆる例外を受け取ることになります。

例えば、途中のどこかの時点で、別の依存関係やpath operationの中で、データベーストランザクションを「ロールバック」したり、その他の例外を作成したりするコードがあった場合、依存関係の中で例外を受け取ることになります。

そのため、依存関係の中にある特定の例外を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

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

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 は、全てが正しい順序で実行されていることを確認します。

技術詳細

これはPythonのContext Managersのおかげで動作します。

FastAPI はこれを実現するために内部的に使用しています。

yieldとHTTPExceptionを持つ依存関係

yieldを持つ依存関係を使い、何らかのコードを実行し、その後にfinallyの後で終了コードを実行しようとするtryブロックを持てることが分かりました。

また、exceptを使って発生した例外をキャッチし、それに対して何かをすることもできます。

例えば、HTTPExceptionのように別の例外を発生させることができます。

豆知識

これはやや高度なテクニックで、ほとんどの場合は本当に必要にはなりません。例えば、path operation 関数など、アプリケーションコードの他の場所から（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

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

例外をキャッチして、それに基づいてカスタムレスポンスを作成したい場合は、カスタム例外ハンドラを作成してください。

yieldとexceptを持つ依存関係

yieldを持つ依存関係でexceptを使って例外をキャッチし、それを再度raiseしない（または新しい例外をraiseしない）場合、通常のPythonと同じように、FastAPIは例外があったことに気づけません:

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

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

この場合、（HTTPExceptionやそれに類するものをraiseしていないため）クライアントには適切にHTTP 500 Internal Server Errorレスポンスが返りますが、サーバーにはログが一切残らず、何がエラーだったのかを示す他の手がかりもありません。 😱

yieldとexceptを持つ依存関係では常にraiseする

yieldを持つ依存関係で例外をキャッチした場合、別のHTTPExceptionなどをraiseするのでない限り、元の例外を再raiseすべきです。

raiseを使うと同じ例外を再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

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 Internal Server Errorレスポンスを受け取りますが、サーバーのログにはカスタムの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

情報

１つのレスポンス だけがクライアントに送信されます。それはエラーレスポンスの一つかもしれませんし、path operationからのレスポンスかもしれません。

いずれかのレスポンスが送信された後、他のレスポンスを送信することはできません。

豆知識

path operation 関数のコードで例外をraiseした場合、HTTPExceptionを含め、それはyieldを持つ依存関係に渡されます。ほとんどの場合、その例外が正しく処理されるように、yieldを持つ依存関係から同じ例外、または新しい例外を再raiseしたくなるでしょう。

早期終了とscope

通常、yieldを持つ依存関係の終了コードは、クライアントにレスポンスが送信された後に実行されます。

しかし、path operation 関数からreturnした後に依存関係を使う必要がないと分かっている場合は、Depends(scope="function")を使って、レスポンスが送信される前に、path operation 関数のreturn後に依存関係を閉じるべきだとFastAPIに伝えられます。

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


def get_username():
    try:
        yield "Rick"
    finally:
        print("Cleanup up before response is sent")


@app.get("/users/me")
def get_user_me(username: Annotated[str, Depends(get_username, scope="function")]):
    return username
🤓 Other versions and variants

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI

app = FastAPI()


def get_username():
    try:
        yield "Rick"
    finally:
        print("Cleanup up before response is sent")


@app.get("/users/me")
def get_user_me(username: str = Depends(get_username, scope="function")):
    return username

Depends()は、以下のいずれかを取るscopeパラメータを受け取ります:

  • "function": リクエストを処理するpath operation 関数の前に依存関係を開始し、path operation 関数の終了後に依存関係を終了しますが、クライアントにレスポンスが返される前に終了します。つまり、依存関係関数はpath operation 関数の周囲で実行されます。
  • "request": リクエストを処理するpath operation 関数の前に依存関係を開始し（"function"を使用する場合と同様）、クライアントにレスポンスが返された後に終了します。つまり、依存関係関数はリクエストとレスポンスのサイクルの周囲で実行されます。

指定されておらず、依存関係にyieldがある場合、デフォルトでscopeは"request"になります。

サブ依存関係のscope

scope="request"（デフォルト）を持つ依存関係を宣言する場合、どのサブ依存関係も"request"のscopeを持つ必要があります。

しかし、"function"のscopeを持つ依存関係は、"function"と"request"のscopeを持つ依存関係を持てます。

これは、いずれの依存関係も、サブ依存関係より前に終了コードを実行できる必要があるためです（終了コードの実行中にサブ依存関係をまだ使う必要がある可能性があるためです）。

sequenceDiagram

participant client as Client
participant dep_req as Dep scope="request"
participant dep_func as Dep scope="function"
participant operation as Path Operation

    client ->> dep_req: Start request
    Note over dep_req: Run code up to yield
    dep_req ->> dep_func: Pass dependency
    Note over dep_func: Run code up to yield
    dep_func ->> operation: Run path operation with dependency
    operation ->> dep_func: Return from path operation
    Note over dep_func: Run code after yield
    Note over dep_func: ✅ Dependency closed
    dep_func ->> client: Send response to client
    Note over client: Response sent
    Note over dep_req: Run code after yield
    Note over dep_req: ✅ Dependency closed

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を持つ依存関係でのコンテキストマネージャの使用

注意

これは多かれ少なかれ、「高度な」発想です。

FastAPI を使い始めたばかりの方は、とりあえずスキップした方がよいかもしれません。

Pythonでは、以下の２つのメソッドを持つクラスを作成する: __enter__()と__exit__()ことでコンテキストマネージャを作成することができます。

また、依存関数の中でwithやasync with文を使用することによってyieldを持つ FastAPI の依存関係の中でそれらを使用することができます:

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

豆知識

コンテキストマネージャを作成するもう一つの方法はwithです:

これらを使って、関数を単一のyieldでデコレートすることができます。

これは FastAPI が内部的にyieldを持つ依存関係のために使用しているものです。

しかし、FastAPIの依存関係にデコレータを使う必要はありません（そして使うべきではありません）。

FastAPIが内部的にやってくれます。