コンテンツへスキップ

依存関係

FastAPIは非常に強力でありながら直感的な依存関係注入システムを備えています。

非常に使いやすいように設計されており、開発者が他のコンポーネントをFastAPIと簡単に統合できるようにしています。

「依存関係注入」とは何か

プログラミングにおいて「依存関係注入」とは、コード(この場合はパスオペレーション関数)が動作および使用するのに必要なものを宣言する方法です。「依存関係」です。

そして、そのシステム(この場合はFastAPI)は、コードに必要な依存関係を提供するために必要な処理を行います(依存関係を「注入」します)。

これは、次のような場合に非常に役立ちます。

  • 共有ロジック(同じコードロジックを繰り返し使用する)。
  • データベース接続を共有する。
  • セキュリティ、認証、ロール要件などを強制する。
  • その他多くのこと…

これらはすべて、コードの重複を最小限に抑えながら実現できます。

最初のステップ

非常に簡単な例を見てみましょう。現時点ではあまり役に立ちませんが。

しかし、これにより、依存関係注入システムの動作に集中できます。

依存関係、または「依存可能」なものを作成する

まず、依存関係に注目しましょう。

それは、パスオペレーション関数と同じパラメータをすべて受け取ることができる関数です。

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Annotated, Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

ヒント

可能であれば、Annotatedバージョンを使用することをお勧めします。

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

ヒント

可能であれば、Annotatedバージョンを使用することをお勧めします。

from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

以上です。

2行です。.

そして、それはすべてのパスオペレーション関数と同じ形状と構造を持っています。

「デコレータ」(@app.get("/some-path"))のないパスオペレーション関数と考えてください。

そして、何でも返すことができます。

この場合、この依存関係は次のことを期待しています。

  • オプションのクエリパラメータqstr型)。
  • オプションのクエリパラメータskipint型)、デフォルトは0
  • オプションのクエリパラメータlimitint型)、デフォルトは100

そして、それらの値を含むdictを返します。

情報

FastAPIはバージョン0.95.0でAnnotatedのサポートを追加し(そして推奨し始めました)。

古いバージョンを使用している場合、Annotatedを使用しようとするとエラーが発生します。

Annotatedを使用する前に、FastAPIのバージョンを少なくとも0.95.1にアップグレードしてください。

Dependsをインポートする

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Annotated, Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

ヒント

可能であれば、Annotatedバージョンを使用することをお勧めします。

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

ヒント

可能であれば、Annotatedバージョンを使用することをお勧めします。

from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

依存関係を「依存先」で宣言する

パスオペレーション関数のパラメータでBodyQueryなどを扱うのと同じように、新しいパラメータでDependsを使用します。

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Annotated, Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

ヒント

可能であれば、Annotatedバージョンを使用することをお勧めします。

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

ヒント

可能であれば、Annotatedバージョンを使用することをお勧めします。

from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

Dependsは、関数のパラメータでBodyQueryなどと同じように使用しますが、動作が少し異なります。

Dependsには、単一のパラメータのみを渡します。

このパラメータは、関数のようなものでなければなりません。

それを直接呼び出さないでください(最後に括弧を追加しないでください)。Depends()にパラメータとして渡すだけです。

そして、その関数は、パスオペレーション関数と同じ方法でパラメータを受け取ります。

ヒント

関数以外にも、依存関係として使用できる「もの」については、次の章で説明します。

新しいリクエストが到着するたびに、**FastAPI**は

  • 正しいパラメータで依存関係(依存可能)関数を呼び出します。
  • 関数から結果を取得します。
  • その結果をパスオペレーション関数のパラメータに代入します。
graph TB

common_parameters(["common_parameters"])
read_items["/items/"]
read_users["/users/"]

common_parameters --> read_items
common_parameters --> read_users

このようにして、共有コードを一度記述するだけで、**FastAPI**がパスオペレーションのためにそれを呼び出す処理を行います。

確認

特別なクラスを作成して**FastAPI**にどこかに渡して「登録」する必要がないことに注意してください。

Dependsに渡すだけで、**FastAPI**がそれ以降の処理を行います。

Annotated依存関係の共有

上記の例では、少しの**コードの重複**があることがわかります。

common_parameters()依存関係を使用する必要がある場合、型注釈とDepends()を付けたパラメータ全体を記述する必要があります。

commons: Annotated[dict, Depends(common_parameters)]

しかし、Annotatedを使用しているため、そのAnnotated値を変数に格納し、複数の場所でそれを利用できます。

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons
from typing import Annotated, Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons
from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons

ヒント

これは標準的なPythonで、「型エイリアス」と呼ばれ、実際には**FastAPI**固有のものではありません。

しかし、**FastAPI**はAnnotatedを含むPython標準に基づいているため、このテクニックをコードで使用できます。😎

依存関係は期待通りに動作し続け、最も重要なことは、型情報が保持されるということです。つまり、エディタは引き続きオートコンプリートインラインエラーなどを提供できます。mypyなどの他のツールについても同様です。

これは、多くのパスオペレーション同じ依存関係を繰り返し使用する場合、特に大規模なコードベースで使用すると非常に便利です。

asyncにするか、しないか

依存関係も**FastAPI**によって呼び出されるため(パスオペレーション関数と同じ)、関数を定義する際に同じ規則が適用されます。

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

通常のdefパスオペレーション関数内でasync defを使用して依存関係を宣言したり、async defパスオペレーション関数内でdef依存関係を宣言したりできます。

問題ありません。**FastAPI**は適切な処理を行います。

注記

わからない場合は、ドキュメントの非同期:「急いでいますか?」セクションでasyncawaitについて確認してください。

OpenAPIとの統合

依存関係(およびサブ依存関係)のリクエスト宣言、検証、要件はすべて、同じOpenAPIスキーマに統合されます。

そのため、インタラクティブなドキュメントにもこれらの依存関係からのすべての情報が表示されます。

簡単な使用方法

パスオペレーション関数は、パスとオペレーションが一致したときに使用されるように宣言され、**FastAPI**はリクエストからデータを取り出して、正しいパラメータで関数を呼び出す処理を行います。

実際、ほとんど(またはすべての)Webフレームワークは同じ方法で動作します。

これらの関数を直接呼び出すことはありません。フレームワーク(この場合は**FastAPI**)によって呼び出されます。

依存関係注入システムを使用すると、パスオペレーション関数もパスオペレーション関数の実行前に実行されるべき他のものに「依存」していることを**FastAPI**に伝えることができ、**FastAPI**はそれを実行し、結果を「注入」します。

この「依存関係注入」の同じ概念に対する他の一般的な用語は次のとおりです。

  • リソース
  • プロバイダー
  • サービス
  • インジェクタブル
  • コンポーネント

**FastAPI**プラグイン

統合と「プラグイン」は、**依存関係注入**システムを使用して構築できます。しかし実際には、依存関係を使用することで、パスオペレーション関数で使用できる無限の数の統合とインタラクションを宣言できるため、**「プラグイン」を作成する必要はありません**。

そして、依存関係は非常にシンプルで直感的な方法で作成できるため、必要なPythonパッケージをインポートし、数行のコードで(文字通り)API関数と統合できます。

関係データベースやNoSQLデータベース、セキュリティなどについて、次の章でその例を示します。

**FastAPI**の互換性

依存関係注入システムのシンプルさにより、**FastAPI**は次のものと互換性があります。

  • すべての関係データベース
  • NoSQLデータベース
  • 外部パッケージ
  • 外部API
  • 認証および承認システム
  • API使用状況監視システム
  • レスポンスデータ注入システム
  • など

シンプルで強力

階層的な依存関係注入システムは定義と使用が非常に簡単ですが、非常に強力です。

依存関係を定義し、それがさらに依存関係を定義できます。

最終的に、依存関係の階層ツリーが構築され、**依存関係注入**システムはこれらの依存関係(およびそのサブ依存関係)をすべて解決し、各ステップで結果を提供(注入)します。

たとえば、4つのAPIエンドポイント(パスオペレーション)があるとします。

  • /items/public/
  • /items/private/
  • /users/{user_id}/activate
  • /items/pro/

それぞれの異なる権限要件を、依存関係とサブ依存関係だけで追加できます。

graph TB

current_user(["current_user"])
active_user(["active_user"])
admin_user(["admin_user"])
paying_user(["paying_user"])

public["/items/public/"]
private["/items/private/"]
activate_user["/users/{user_id}/activate"]
pro_items["/items/pro/"]

current_user --> active_user
active_user --> admin_user
active_user --> paying_user

current_user --> public
active_user --> private
admin_user --> activate_user
paying_user --> pro_items

**OpenAPI**との統合

これらの依存関係は、要件を宣言する一方で、パスオペレーションにパラメータ、検証などを追加します。

**FastAPI**は、それをすべてOpenAPIスキーマに追加するため、インタラクティブなドキュメントシステムに表示されます。