依存性注入¶
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
「依存する側」で依存性を宣言する¶
Body
、Query
などと同様に、パス操作関数のパラメータで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
は、Body
、Query
などと同じように関数のパラメータで使用しますが、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に固有のものではありません。
しかし、FastAPIはAnnotated
を含むPython標準に基づいているため、このテクニックをコードで使用できます。😎
依存性は期待どおりに機能し続け、最も良い点は型情報が保持されることです。つまり、エディタは引き続きオートコンプリート、インラインエラーなどを提供できます。mypy
などの他のツールも同様です。
これは、大規模なコードベースで同じ依存関係を多数のパス操作で繰り返し使用する場合に特に役立ちます。
async
にするか、しないか¶
依存関係もFastAPIによって呼び出されるため(パス操作関数と同様)、関数を定義する際には同じルールが適用されます。
async def
または通常のdef
を使用できます。
そして、通常のdef
パス操作関数内でasync def
依存関係を宣言したり、async def
パス操作関数内でdef
依存関係を宣言したりできます。
どちらでも構いません。FastAPIはどのように処理すべきかを知っています。
メモ
ご存じない場合は、ドキュメントのasync
とawait
に関する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スキーマに追加し、インタラクティブなドキュメントシステムに表示されるようにします。