Zenn
📘

FastAPIでDIをする (Dependency Injectorを使う)

2022/02/23に公開
Python
FastAPI
#
di
tech

FastAPIを使う実装にてDI (Dependency Injection, 依存性注入) を行う方法を書き残します。日々模索しており、理解不足の自覚強いので自信ありません。

記事の概要

  • FastAPIが標準で持つDI機構に加えて、上乗せして Dependency Injector を使うことで見通し良く実装できます。
  • FastAPI標準では対処しづらい、SingletonなDI (ただ1つのオブジェクトを常に注入する) もDependency Injector 合わせ技により簡単に実現できます。

作成したサンプルプログラムです。
https://github.com/shimat/fastapi_di_sample

FastAPI標準のDIと、その課題

FastAPIは標準でDIシステムを持っており、まずはそれを使うのが基本です。FastAPIはドキュメントがとても丁寧ですから本記事では説明は省きます。
https://fastapi.tiangolo.com/tutorial/dependencies/

しかし私がFastAPIで何回か開発するたび、以下のような課題を感じました。

課題1. テストごとに別のモックを注入することがしづらい

以下のStack Overflowと同じ悩みです。
https://stackoverflow.com/questions/65259085/best-way-to-override-fastapi-dependencies-for-testing-with-a-different-dependenc

エンドポイントの定義例

def common_parameters() -> dict[str, str]:
    return {"message": "Hello world"}


@app.get("/hoge")
def get_hoge(params: dict[str, str] = Depends(common_parameters)):
    return { "result": params["message"] }

そして以下は、common_parametersを複数種類のモックで差し替えてテストする例です。コメントしている箇所のような課題が生まれるので、dependency_overridesは都度復元すべきと考えます。

def mock_parameters_1():
    return {"message": "See you"}


def mock_parameters_2():
    return {"message": "Bye bye"}    


client = TestClient(app)

def test_mock_1():
    app.dependency_overrides[common_parameters] = mock_parameters_1
    response = client.get("/hoge")
    assert response.json() == { "result": "See you" }
    

def test_mock_2():
    app.dependency_overrides[common_parameters] = mock_parameters_2
    response = client.get("/hoge")
    assert response.json() == { "result": "Bye bye" }


def test_normal():
    response = client.get("/hoge")
    # このテストでは dependency_overrides を変更していないが、上で破壊されているので失敗する
    assert response.json() == { "result": "Hello world" }

この課題に対し、上のStack Overflowの回答に1つの解決策があります。ただ、案件ごとに毎度のことですし、あまりテストに複雑な実装を入れるのも億劫に感じます（テストのテストが欲しくなってくる）。

課題2. "SingletonのDI"がやりづらい

（用語がおかしいかもしれません。）注入するオブジェクトをただ1つのものにしたいということです。FastAPIのDependsはリクエストの都度呼び出されるため、普通に書くとこの用途に向きません。[1]

# すごく重い初期化処理
def common_parameters() -> dict[str, str]:
    sleep(10)
    return {"message": "Hello world"}


# 毎度 common_parameters が呼び出され、10秒かかる
@app.get("/hoge")
def get_hoge(params: dict[str, str] = Depends(common_parameters)):
    return { "result": params["message"] }

これにまつわるissueがFastAPIのGitHubにあります。[2]
https://github.com/tiangolo/fastapi/issues/504

正直、長くて完全に把握できていませんが、いくつか解決策は提示されています。ただしFastAPIのみでの楽な解決策は無さそうと判断しました。長い独自のラッパー層はあまり書きたくないので、他のDIライブラリと合わせ技を考えました。

Dependency Injector と併用する

PythonのDIライブラリについて

今回は Dependency Injector (python-dependency-injector) を使うことにしました。
https://python-dependency-injector.ets-labs.org/index.html

PythonのDIライブラリは星の数ほどあるようで、例をあげれば python-inject や Injector などがあります。比較して選んだわけではありません。単に、FastAPIと併用するドキュメント・サンプルが公式で準備されていたのを最初に見つけたのが理由です。

正直なところ、このドキュメントに要点が揃っていてそれ以上書くこともなく、ここではあまり説明はしないことにします。

ほかにboto3と併用するサンプル等もありまして、これらを組み合わせていくと、AWS等のクラウド上で動作しS3、RDS、SQS、MLのモデル、ローカルの巨大なCSV、等々・・・といったリソースに触るクラスをDIするようなFastAPIを作るのに良い参考になると思いました。

Dependency Injectorと組み合わせる際の要点

以下はDependency Injector公式のサンプルと、私が作成したサンプルです。ここから要点だけ以下述べます。

FactoryかSingletonか

一番基本的なのがFactoryプロバイダで、できることは大体FastAPIのDependsと同じと考えています。
https://python-dependency-injector.ets-labs.org/providers/factory.html

プロバイダはほかにもたくさんあり、Singletonもあります。これがはじめに述べた課題にピタリはまります。
https://python-dependency-injector.ets-labs.org/providers/singleton.html

使用例が以下の箇所です。
https://github.com/shimat/fastapi_di_sample/blob/8464f851101d11bd6b4832cb61282a829c2cd8ed/src/fastapi_di_sample/containers.py#L14

containers.py
class Container(containers.DeclarativeContainer):
    wiring_config = containers.WiringConfiguration(modules=[".endpoints"])

    # 一応ThreadSafe版を選んでみました
    my_client = providers.ThreadSafeSingleton(MyDbClient)

    my_service = providers.Factory(
        services.MyService,
        client=my_client,
    )

なお、Resource プロバイダというものもあり、これはSingletonと似ていてかつ initとshutdownをさらにサポートしています。初期化処理・解放処理を書きたい時 (DBのコネクションなど) に向いていそうです。
https://python-dependency-injector.ets-labs.org/providers/resource.html

FastAPIでシングルトンオブジェクトを注入する場合は、スレッドセーフかどうかが重要です。必要に迫られない限りは普通のFactoryのプロバイダの方が良さそうです。例えばboto3は使い方によってはスレッドセーフでない場合があるので、無理にSingletonにしない方が良いと考えます。（参考: https://dev.classmethod.jp/articles/python-boto3-session-thread-unsafe/ ）

テストでモックを注入する

テストをうまく書きたいからこそDIしているようなものです。ここも、はじめに述べた課題を解決できました。
https://github.com/shimat/fastapi_di_sample/blob/main/tests/test_fastapi_di_sample.py

test_fastapi_di_sample.py
def test_get_mock():
    client_mock = mock.MagicMock(spec=IMyDbClient)
    client_mock.get_ok_words.return_value = ["foo", "bar", "baz"]

    # DIのclientをモックで上書き
    with app.container.my_client.override(client_mock):
        request_params = {"word": "bar"}
        response = test_client.get("/", params=request_params)
        assert response.status_code == 200
        assert response.json() == {"is_ok": True}

withの中でだけモックに差し変わり、テストが終われば元に戻ります。副作用が無くなるので、自由に何度もモックテストを記述できます。

mypyに指摘される点

以下の行がmypyに指摘されるはずです。app(FastAPI)はcontainerなんて持ってないぞ、と。
https://github.com/shimat/fastapi_di_sample/blob/8464f851101d11bd6b4832cb61282a829c2cd8ed/src/fastapi_di_sample/app.py#L12

def create_app() -> FastAPI:
    container = Container()

    app = FastAPI()
    # ↓ ここ
    app.container = container  # type: ignore

ここはDependency Injectorのサンプルの書き方を踏襲したまでで、私は警告を抑止する方針にしましたが、気になるのであれば無理にappに持たせなくても実装はできると思います。

脚注

  1. この例で言えば @functools.cache デコレータを付ければ一見して解決します。後述するissueでも何人かこの案を述べています。私はそれで良いのかどうか判断できませんでした。 ↩︎

  2. 巨大なCSVを何度も開きたくないとか、DBのコネクションを維持したい、といった例がissueで挙げられています。 ↩︎

Discussion