yield を使用した依存関係

FastAPI は、終了後にいくつかの追加手順を実行する依存関係をサポートしています。

これを行うには、return の代わりに yield を使用し、その後に追加の手順 (コード) を記述します。

ヒント

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

技術的な詳細

以下のいずれかと一緒に使用できる有効な関数はすべて

• @contextlib.contextmanager または
• @contextlib.asynccontextmanager

は、FastAPI 依存関係として使用できます。

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

yield を使用したデータベース依存関係

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

レスポンスが作成される前に実行されるのは、yield ステートメントまでのコードのみです。

Python 3.8 以降

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

yield された値が、パス操作 およびその他の依存関係に注入される値です。

Python 3.8 以降

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

yield ステートメントに続くコードは、レスポンスの作成後に実行されますが、送信前です。

Python 3.8 以降

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

ヒント

async 関数または通常の関数を使用できます。

FastAPI は、通常の依存関係と同様に、それぞれの関数を適切に処理します。

yield と try を使用した依存関係

yield を使用する依存関係で try ブロックを使用すると、依存関係の使用時にスローされた例外を受け取ることができます。

たとえば、途中のどこかで、別の依存関係やパス操作で、データベーストランザクションが「ロールバック」したり、その他のエラーが発生した場合、依存関係で例外を受け取ります。

したがって、except SomeException を使用して、依存関係内でその特定の例外を探すことができます。

同様に、finally を使用して、例外の有無にかかわらず、終了ステップが確実に実行されるようにすることができます。

Python 3.8 以降

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

yield を使用したサブ依存関係

サブ依存関係と、任意のサイズと形状のサブ依存関係の「ツリー」を持つことができ、それらのいずれかまたはすべてが yield を使用できます。

FastAPI は、yield を使用する各依存関係の「終了コード」が正しい順序で実行されることを保証します。

たとえば、dependency_c は dependency_b に依存し、dependency_b は dependency_a に依存できます。

Python 3.9 以降

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)

🤓 その他のバージョンとバリアント

Python 3.8 以降Python 3.8 以降 - 非アノテーション

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 と名付けられている) の値が利用可能である必要があります。

Python 3.9 以降

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)

🤓 その他のバージョンとバリアント

Python 3.8 以降Python 3.8 以降 - 非アノテーション

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

技術的な詳細

これは、Python の コンテキストマネージャ のおかげで機能します。

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

yield と HTTPException を使用した依存関係

yield を使用する依存関係を使用し、例外をキャッチする try ブロックを持つことができることを見ました。

同様に、yield の後に、終了コードで HTTPException などを発生させることもできます。

ヒント

これはやや高度な手法であり、ほとんどの場合、実際に必要とすることはありません。なぜなら、アプリケーションコードの残りの部分 (たとえば、パス操作関数) から例外 (HTTPException を含む) を発生させることができるからです。

しかし、必要であれば、それはそこにあります。 🤓

Python 3.9 以降

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

🤓 その他のバージョンとバリアント

Python 3.8 以降Python 3.8 以降 - 非アノテーション

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 を使用して例外をキャッチし、それを再度発生させない場合 (または新しい例外を発生させない場合)、通常の Python と同じように、FastAPI は例外があったことに気付くことができません。

Python 3.9 以降

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

🤓 その他のバージョンとバリアント

Python 3.8 以降Python 3.8 以降 - 非アノテーション

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 を使用して同じ例外を再発生させることができます。

Python 3.9 以降

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

🤓 その他のバージョンとバリアント

Python 3.8 以降Python 3.8 以降 - 非アノテーション

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 つであるか、パス操作からのレスポンスである可能性があります。

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

ヒント

この図は HTTPException を示していますが、yield を使用する依存関係またはカスタム例外ハンドラでキャッチする他の例外も発生させることができます。

例外を発生させると、HTTPException を含む yield を持つ依存関係に渡されます。ほとんどの場合、適切に処理されるように、その同じ例外または yield を持つ依存関係からの新しい例外を再発生させたいと思うでしょう。

yield、HTTPException、except およびバックグラウンドタスクを使用した依存関係

警告

これらの技術的な詳細は、おそらく必要ありません。このセクションをスキップして、以下に進むことができます。

これらの詳細は、FastAPI の 0.106.0 より前のバージョンを使用しており、バックグラウンドタスクで yield を使用する依存関係からリソースを使用していた場合に主に役立ちます。

yield と except を使用した依存関係、技術的な詳細

FastAPI 0.110.0 より前は、yield を使用する依存関係を使用し、その依存関係で except で例外をキャッチし、例外を再度発生させなかった場合、例外は例外ハンドラーまたは内部サーバーエラーハンドラーに自動的に発生/転送されていました。

これは、ハンドラーなしで転送された例外 (内部サーバーエラー) からの処理されていないメモリ消費を修正し、通常の Python コードの動作と一貫性を持たせるために、バージョン 0.110.0 で変更されました。

バックグラウンドタスクと 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 ステートメントを使用して、yield を使用する FastAPI 依存関係内で使用することもできます。

Python 3.8 以降

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

ヒント

コンテキストマネージャを作成する別の方法は

• @contextlib.contextmanager または
• @contextlib.asynccontextmanager

単一の yield を持つ関数をデコレートするためにそれらを使用します。

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

しかし、FastAPI の依存関係にデコレータを使用する必要はありません (また、使用すべきではありません)。

FastAPI が内部的に処理します。