コンテンツにスキップ

依存関係 - 最初のステップ

** FastAPI** は非常に強力でありながら直感的な 依存性注入 システムを持っています。

それは非常にシンプルに使用できるように設計されており、開発者が他のコンポーネント FastAPI と統合するのが非常に簡単になるように設計されています。

「依存性注入」とは

「依存性注入」 とは、プログラミングにおいて、コード(この場合は、path operation関数)が動作したり使用したりするために必要なもの(「依存関係」)を宣言する方法があることを意味します:

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

これは以下のようなことが必要な時にとても便利です:

  • ロジックを共有している。(同じコードロジックを何度も繰り返している)。
  • データベース接続を共有する。
  • セキュリティ、認証、ロール要件などを強制する。
  • そのほかにも多くのこと...

これらすべてを、コードの繰り返しを最小限に抑えながら行います。

最初のステップ

非常にシンプルな例を見てみましょう。あまりにもシンプルなので、今のところはあまり参考にならないでしょう。

しかし、この方法では 依存性注入 システムがどのように機能するかに焦点を当てることができます。

依存関係の作成

まずは依存関係に注目してみましょう。

以下のように、path operation関数と同じパラメータを全て取ることができる関数にすぎません:

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
🤓 Other versions and variants
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

Tip

Prefer to use the Annotated version if possible.

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

これだけです。

2行

そして、それはすべてのpath operation関数が持っているのと同じ形と構造を持っています。

「デコレータ」を含まない(@app.get("/some-path")を含まない)path operation関数と考えることもできます。

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

この場合、この依存関係は以下を期待しています:

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

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

Dependsのインポート

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
🤓 Other versions and variants
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

Tip

Prefer to use the Annotated version if possible.

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

"dependant"での依存関係の宣言

path operation関数のパラメータにBodyQueryなどを使用するのと同じように、新しいパラメータにDependsを使用することができます:

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
🤓 Other versions and variants
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

Tip

Prefer to use the Annotated version if possible.

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

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

Dependsは1つのパラメータしか与えられません。

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

そして、その関数は、path operation関数が行うのと同じ方法でパラメータを取ります。

豆知識

次の章では、関数以外の「もの」が依存関係として使用できるものを見ていきます。

新しいリクエストが到着するたびに、FastAPI が以下のような処理を行います:

  • 依存関係("dependable")関数を正しいパラメータで呼び出します。
  • 関数の結果を取得します。
  • path operation関数のパラメータにその結果を代入してください。
graph TB

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

common_parameters --> read_items
common_parameters --> read_users

この方法では、共有されるコードを一度書き、FastAPIpath operationsのための呼び出しを行います。

確認

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

Dependsを渡すだけで、FastAPI が残りの処理をしてくれます。

asyncにするかどうか

依存関係は FastAPIpath operation関数と同じ)からも呼び出されるため、関数を定義する際にも同じルールが適用されます。

async defや通常のdefを使用することができます。

また、通常のdefpath operation関数の中にasync defを入れて依存関係を宣言したり、async defpath operation関数の中にdefを入れて依存関係を宣言したりすることなどができます。

それは重要ではありません。FastAPI は何をすべきかを知っています。

備考

わからない場合は、ドキュメントのAsync: "In a hurry?"の中のasyncawaitについてのセクションを確認してください。

OpenAPIとの統合

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

つまり、対話型ドキュメントにはこれらの依存関係から得られる全ての情報も含まれているということです:

簡単な使い方

見てみると、pathoperationが一致した時にpath operation関数が宣言されていて、FastAPI が正しいパラメータで関数を呼び出してリクエストからデータを抽出する処理をしています。

実は、すべての(あるいはほとんどの)Webフレームワークは、このように動作します。

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

依存性注入システムでは、FastAPIpath operationもまた、path operation関数の前に実行されるべき他の何かに「依存」していることを伝えることができ、FastAPI がそれを実行し、結果を「注入」することを引き受けます。

他にも、「依存性注入」と同じような考えの一般的な用語があります:

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

FastAPI プラグイン

統合や「プラグイン」は 依存性注入 システムを使って構築することができます。しかし、実際には、「プラグイン」を作成する必要はありません。依存関係を使用することで、無限の数の統合やインタラクションを宣言することができ、それが*path operation関数で利用可能になるからです。

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

次の章では、リレーショナルデータベースやNoSQLデータベース、セキュリティなどについて、その例を見ていきます。

FastAPI 互換性

依存性注入システムがシンプルなので、FastAPI は以下のようなものと互換性があります:

  • すべてのリレーショナルデータベース
  • NoSQLデータベース
  • 外部パッケージ
  • 外部API
  • 認証・認可システム
  • API利用状況監視システム
  • レスポンスデータ注入システム
  • など。

シンプルでパワフル

階層依存性注入システムは、定義や使用方法が非常にシンプルであるにもかかわらず、非常に強力なものとなっています。

依存関係事態を定義する依存関係を定義することができます。

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

例えば、4つのAPIエンドポイント(path operations)があるとします:

  • /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 との統合

これら全ての依存関係は、要件を宣言すると同時に、path operationsにパラメータやバリデーションを追加します。

FastAPI はそれをすべてOpenAPIスキーマに追加して、対話型のドキュメントシステムに表示されるようにします。