高度な依存関係

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

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

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

クエリパラメーター q が特定の固定コンテンツを含むかどうかをチェックする依存関係があると想像してみましょう。

しかし、その固定コンテンツをパラメーター化できるようにしたいのです。

"呼び出し可能"なインスタンス

Pythonでは、クラスのインスタンスを「呼び出し可能」にする方法があります。

クラス自体（すでに呼び出し可能です）ではなく、そのクラスのインスタンスです。

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

Python 3.9+

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}

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

Python 3.8+Python 3.8+ - non-Annotated

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

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}

ヒント

可能であれば`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: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

この場合、この __call__ は、FastAPI が追加のパラメーターとサブ依存関係をチェックするために使用するものであり、後でパス操作関数のパラメーターに値を渡すために呼び出されるものです。

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

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

Python 3.9+

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}

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

Python 3.8+Python 3.8+ - non-Annotated

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

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}

ヒント

可能であれば`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: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

この場合、FastAPI は __init__ に触れたり、気にしたりすることはなく、コードで直接使用します。

インスタンスの作成

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

Python 3.9+

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}

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

Python 3.8+Python 3.8+ - non-Annotated

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

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}

ヒント

可能であれば`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: 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")

...そして、それが返すものをすべて、パス操作関数の依存関係の値として、パラメーター fixed_content_included として渡します。

Python 3.9+

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}

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

Python 3.8+Python 3.8+ - non-Annotated

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

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}

ヒント

可能であれば`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: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

ヒント

これらすべては不自然に見えるかもしれません。そして、それがまだどのように役立つかはあまり明確ではないかもしれません。

これらの例は意図的に単純ですが、すべてがどのように機能するかを示しています。

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

これらすべてを理解していれば、セキュリティのためのそれらのユーティリティツールが内部でどのように機能するかをすでに知っています。

yield、HTTPException、except、およびバックグラウンドタスクを持つ依存関係

Warning

おそらく、これらの技術的な詳細は必要ないでしょう。

これらの詳細は、主にFastAPIアプリケーションが0.118.0より古く、yieldを持つ依存関係で問題に直面している場合に役立ちます。

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

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

FastAPI 0.118.0より前は、yield を伴う依存関係を使用した場合、パス操作関数が返された後、レスポンスを送信する直前に終了コードが実行されていました。

これは、リソースを必要以上に長く保持し、レスポンスがネットワークを介して送信されるのを待つことを避けるためでした。

この変更は、StreamingResponse を返した場合、yield を伴う依存関係の終了コードがすでに実行されていたことも意味しました。

例えば、yield を伴う依存関係にデータベースセッションがあった場合、StreamingResponse はデータのストリーミング中にそのセッションを使用できませんでした。なぜなら、セッションは yield の後の終了コードですでに閉じられていたからです。

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

情報

以下に示すように、これはバージョン0.106.0以前の動作と非常によく似ていますが、いくつかの改善点と特殊ケースのバグ修正が含まれています。

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

特定の条件を持ついくつかのユースケースでは、レスポンスを送信する前に yield を持つ依存関係の終了コードを実行する古い動作がメリットをもたらす可能性があります。

例えば、ユーザーを検証するためだけに yield を持つ依存関係でデータベースセッションを使用するコードがあり、そのデータベースセッションがパス操作関数で二度と使用されず、依存関係内でのみ使用される、さらにレスポンスの送信に時間がかかる場合（例えば、データをゆっくり送信する 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 👇

👀 ファイル全体のプレビュー

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))

👀 ファイル全体のプレビュー

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 👇

👀 ファイル全体のプレビュー

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ディスカッションの質問を作成してください。

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を渡し、バックグラウンドタスク関数内で再度オブジェクトを取得します。