高度な依存関係

🌐 AI と人間による翻訳

この翻訳は、人間のガイドに基づいて AI によって作成されました。🤝

原文の意図を取り違えていたり、不自然な表現になっている可能性があります。🤖

AI LLM をより適切に誘導するのを手伝う ことで、この翻訳を改善できます。

英語版

パラメータ化された依存関係

これまで見てきた依存関係は、固定の関数またはクラスでした。

しかし、多くの異なる関数やクラスを宣言せずに、その依存関係にパラメータを設定したい場合があります。

たとえば、クエリパラメータ q に、ある固定の内容が含まれているかを検査する依存関係が欲しいとします。

ただし、その固定の内容はパラメータ化できるようにしたいです。

"callable" なインスタンス

Python には、クラスのインスタンスを "callable" にする方法があります。

クラス自体（これはすでに callable です）ではなく、そのクラスのインスタンスです。

そのためには、__call__ メソッドを宣言します:

Python 3.10+

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

🤓 Other versions and variants

Python 3.9+Python 3.10+ - non-AnnotatedPython 3.9+ - non-Annotated

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

この場合、この __call__ が、FastAPI が追加のパラメータやサブ依存関係を確認するために使うものになり、後であなたの path operation 関数 のパラメータに値を渡すために呼び出されるものになります。

インスタンスのパラメータ化

そして、__init__ を使って、依存関係を「パラメータ化」するために利用できるインスタンスのパラメータを宣言できます:

Python 3.10+

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

🤓 Other versions and variants

Python 3.9+Python 3.10+ - non-AnnotatedPython 3.9+ - non-Annotated

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

この場合、FastAPI は __init__ に触れたり気にかけたりすることはありません。私たちがコード内で直接使います。

インスタンスの作成

このクラスのインスタンスは次のように作成できます:

Python 3.10+

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

🤓 Other versions and variants

Python 3.9+Python 3.10+ - non-AnnotatedPython 3.9+ - non-Annotated

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

このようにして依存関係を「パラメータ化」できます。いまや "bar" が属性 checker.fixed_content として中に保持されています。

インスタンスを依存関係として使う

その後、Depends(FixedContentQueryChecker) の代わりに Depends(checker) でこの checker を使えます。依存関係はクラスそのものではなく、インスタンスである checker だからです。

依存関係を解決するとき、FastAPI はこの checker を次のように呼び出します:

checker(q="somequery")

...そして、その戻り値を path operation 関数 内の依存関係の値として、パラメータ fixed_content_included に渡します:

Python 3.10+

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

🤓 Other versions and variants

Python 3.9+Python 3.10+ - non-AnnotatedPython 3.9+ - non-Annotated

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

豆知識

ここまでの内容は回りくどく感じられるかもしれません。まだどのように役立つかが明確でないかもしれません。

これらの例は意図的に単純ですが、仕組みを示しています。

セキュリティの章では、同じやり方で実装されたユーティリティ関数があります。

ここまでを理解できていれば、そうしたセキュリティ用ユーティリティが内部でどのように動いているかも理解できています。

yield、HTTPException、except とバックグラウンドタスクを伴う依存関係

注意

これらの技術的詳細は、ほとんどの場合は不要です。

主に、0.121.0 より前の FastAPI アプリケーションがあり、yield を使う依存関係で問題が発生している場合に有用です。

yield を使う依存関係は、さまざまなユースケースに対応し、いくつかの問題を修正するために時間とともに進化してきました。ここでは変更点の概要を説明します。

yield と scope を伴う依存関係

バージョン 0.121.0 で、yield を使う依存関係に対して Depends(scope="function") がサポートされました。

Depends(scope="function") を使うと、yield の後の終了コードは、クライアントへレスポンスが返される前、path operation 関数 が終了した直後に実行されます。

そして、Depends(scope="request")（デフォルト）を使う場合、yield の後の終了コードはレスポンス送信後に実行されます。

詳しくはドキュメント「yield を使う依存関係 - 早期終了と scope」を参照してください。

yield と StreamingResponse を伴う依存関係、技術詳細

FastAPI 0.118.0 より前では、yield を使う依存関係を使用すると、path operation 関数 が戻ってからレスポンス送信直前に終了コードが実行されていました。

これは、レスポンスがネットワーク上を移動するのを待っている間に、不要にリソースを保持しないようにする意図でした。

この変更により、StreamingResponse を返す場合、yield を持つ依存関係の終了コードはすでに実行されていることになりました。

たとえば、yield を持つ依存関係の中でデータベースセッションを持っていた場合、StreamingResponse はデータをストリーミングしている間にそのセッションを使えません。というのも、yield の後の終了コードでそのセッションがすでにクローズされているからです。

この挙動は 0.118.0 で元に戻され、yield の後の終了コードはレスポンス送信後に実行されるようになりました。

情報

以下で見るように、これはバージョン 0.106.0 より前の挙動ととても似ていますが、いくつかのコーナーケースに対する改良とバグ修正が含まれています。

早期終了コードのユースケース

特定の条件では、レスポンス送信前に yield を持つ依存関係の終了コードを実行する、古い挙動の恩恵を受けられるユースケースがあります。

例えば、yield を持つ依存関係でデータベースセッションを使ってユーザ検証だけを行い、その後は path operation 関数 内ではそのデータベースセッションを一切使わない、かつレスポンス送信に長い時間がかかる（例えばデータをゆっくり送る StreamingResponse）が、何らかの理由でデータベースは使わない、というケースです。

この場合、レスポンスの送信が終わるまでデータベースセッションが保持されますが、使わないのであれば保持する必要はありません。

次のようになります:

Python 3.10+

import time
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine

engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str


app = FastAPI()


def get_session():
    with Session(engine) as session:
        yield session


def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=403, detail="Not authorized")


def generate_stream(query: str):
    for ch in query:
        yield ch
        time.sleep(0.1)


@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
    return StreamingResponse(content=generate_stream(query))

終了コード、すなわち Session の自動クローズは:

Python 3.10+

# Code above omitted 👆

def get_session():
    with Session(engine) as session:
        yield session

# Code below omitted 👇

👀 Full file preview

Python 3.10+

import time
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine

engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str


app = FastAPI()


def get_session():
    with Session(engine) as session:
        yield session


def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=403, detail="Not authorized")


def generate_stream(query: str):
    for ch in query:
        yield ch
        time.sleep(0.1)


@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
    return StreamingResponse(content=generate_stream(query))

...の部分で定義されており、遅いデータ送信が終わった後に実行されます:

Python 3.10+

# Code above omitted 👆

def generate_stream(query: str):
    for ch in query:
        yield ch
        time.sleep(0.1)


@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
    return StreamingResponse(content=generate_stream(query))

👀 Full file preview

Python 3.10+

import time
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine

engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str


app = FastAPI()


def get_session():
    with Session(engine) as session:
        yield session


def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=403, detail="Not authorized")


def generate_stream(query: str):
    for ch in query:
        yield ch
        time.sleep(0.1)


@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
    return StreamingResponse(content=generate_stream(query))

しかし、generate_stream() はデータベースセッションを使わないため、レスポンス送信中にセッションを開いたままにしておく必要は実際にはありません。

SQLModel（または SQLAlchemy）でこの特定のユースケースがある場合は、不要になった時点でセッションを明示的にクローズできます:

Python 3.10+

# Code above omitted 👆

def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=403, detail="Not authorized")
    session.close()

# Code below omitted 👇

👀 Full file preview

Python 3.10+

import time
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine

engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str


app = FastAPI()


def get_session():
    with Session(engine) as session:
        yield session


def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=403, detail="Not authorized")
    session.close()


def generate_stream(query: str):
    for ch in query:
        yield ch
        time.sleep(0.1)


@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
    return StreamingResponse(content=generate_stream(query))

このようにすると、セッションはデータベース接続を解放するため、他のリクエストがそれを使えるようになります。

yield を持つ依存関係で早期終了が必要な別のユースケースがある場合は、あなたの具体的なユースケースと、なぜ yield を持つ依存関係の早期クローズが有益かを説明して、GitHub Discussion の質問を作成してください。

yield を持つ依存関係の早期クローズに納得できるユースケースがある場合は、早期クローズにオプトインする新しい方法を追加することを検討します。

yield と except を伴う依存関係、技術詳細

FastAPI 0.110.0 より前では、yield を持つ依存関係を使い、その依存関係内で except によって例外を捕捉し、再度その例外を送出しなかった場合でも、その例外は自動的に送出（フォワード）され、任意の例外ハンドラまたは内部サーバエラーハンドラに渡されていました。

これは、ハンドラのないフォワードされた例外（内部サーバエラー）による未処理のメモリ消費を修正し、通常の Python コードの挙動と一貫性を持たせるため、バージョン 0.110.0 で変更されました。

バックグラウンドタスクと yield を伴う依存関係、技術詳細

FastAPI 0.106.0 より前では、yield の後で例外を送出することはできませんでした。yield を持つ依存関係の終了コードはレスポンス送信「後」に実行されるため、例外ハンドラ はすでに実行済みでした。

これは主に、依存関係が "yield" した同じオブジェクトをバックグラウンドタスク内で利用できるようにするための設計でした。終了コードはバックグラウンドタスク完了後に実行されるからです。

これは、レスポンスがネットワーク上を移動するのを待っている間にリソースを保持しないようにする意図で、FastAPI 0.106.0 で変更されました。

豆知識

加えて、バックグラウンドタスクは通常、独立したロジックの集合であり、（例えば専用のデータベース接続など）それ自身のリソースで個別に扱うべきです。

そのため、このやり方の方がコードはおそらくよりクリーンになります。

この挙動に依存していた場合は、バックグラウンドタスク用のリソースをバックグラウンドタスク内部で作成し、yield を持つ依存関係のリソースに依存しないデータだけを内部で使用するようにしてください。

例えば、同じデータベースセッションを使うのではなく、バックグラウンドタスク内で新しいデータベースセッションを作成し、この新しいセッションでデータベースからオブジェクトを取得します。そして、バックグラウンドタスク関数の引数としてデータベースのオブジェクト自体を渡すのではなく、そのオブジェクトの ID を渡し、バックグラウンドタスク関数内でもう一度そのオブジェクトを取得します。