コンテンツへスキップ

依存性注入

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")なし)を持たないパス操作関数と考えることができます。

そして、それはあなたが望むどんなものでも返すことができます。

この場合、この依存性は次のものを期待します。

  • strであるオプションのクエリパラメータq
  • intであるオプションのクエリパラメータskip(デフォルトは0)。
  • intであるオプションのクエリパラメータlimit(デフォルトは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には1つのパラメータしか与えません。

このパラメータは関数のようなものである必要があります。

直接呼び出さず(末尾に括弧を追加しない)、単に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に固有のものではありません。

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

依存性は期待どおりに機能し続け、最も良い点型情報が保持されることです。つまり、エディタは引き続きオートコンプリートインラインエラーなどを提供できます。mypyなどの他のツールも同様です。

これは、大規模なコードベース同じ依存関係多数のパス操作で繰り返し使用する場合に特に役立ちます。

asyncにするか、しないか

依存関係もFastAPIによって呼び出されるため(パス操作関数と同様)、関数を定義する際には同じルールが適用されます。

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

そして、通常のdefパス操作関数内でasync def依存関係を宣言したり、async defパス操作関数内でdef依存関係を宣言したりできます。

どちらでも構いません。FastAPIはどのように処理すべきかを知っています。

メモ

ご存じない場合は、ドキュメントのasyncawaitに関するAsync: 「お急ぎですか?」セクションを確認してください。

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スキーマに追加し、インタラクティブなドキュメントシステムに表示されるようにします。