😀

FastAPIとinjectorでリクエストスコープにも対応するDIコンテナ (fastapi-injector)

2025/10/05に公開

はじめに

会社のプロダクト開発でPythonを使ったWebサービスを作ることになり、FastAPIについて学習をしています。
FastAPIを触っていく中で、injectorとfastapi-injectorというライブラリを使うことでDI(Dependency Injection)部分をシンプルな記述で個人的な好みになることがわかったので、記事として書き留めていこうと思います。

FastAPIのDIは以下の方法がありますが、今回は3.に関する記事となります。

1. FastAPI標準のDI
1. python-dependency-injectorライブラリを使用
1. injectorライブラリを使用

FastAPIとinjector

injector自体はDIコンテナのライブラリですがFastAPIと統合されておらず、
HTTPリクエストを考慮したインスタンス生成のスコープは持っていません。
そのため、リクエストスコープのように「同一リクエストでは常に同じインスタンスを取得する」という設定ができません。

シングルトン(Singleton): インスタンスをアプリ全体で1回のみ生成する
リクエストスコープ(Request Scope): 1つのリクエストに対して1つのインスタンスを生成する
トランジェント(Transient): 毎回インスタンスを生成する

将来的に「リクエスト単位で同じインスタンスを使いたい...」となった時にinjectorでは困ってしまうと思いましたが、この問題を解決してくれるfastapi-injectorというライブラリがありました。

インストール

injectorとfastapi-injector(FastAPIとuvicornも含む)をインストールします。

pip install fastapi uvicorn injector fastapi-injector

一応、執筆時点での最新バージョンも記載しておきます。

pip install fastapi==0.118.0 uvicorn==0.37.0 injector==0.22.0 fastapi-injector==0.8.0

使い方

基本的な使い方の全体像は以下となります。

server.py

from fastapi_injector import Injected, InjectorMiddleware, attach_injector, request_scope
from injector import Binder, Injector, Module
import uvicorn
from fastapi import FastAPI

from services.counter_service import CounterService, ICounterService


class AppModule(Module):
    def configure(self, binder: Binder):
        binder.bind(ICounterService, to=CounterService, scope=request_scope)
        return


app = FastAPI()

injector = Injector([AppModule()])
app.add_middleware(InjectorMiddleware, injector=injector)
attach_injector(app=app, injector=injector)


@app.get("/")
def get_root(counter_service: ICounterService = Injected(ICounterService)):
    before = counter_service.get_count()
    counter_service.increment()
    after = counter_service.get_count()

    return {
        "before": before,
        "after": after,
    }


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Moduleを継承したクラスのconfigureメソッドでインジェクションの設定を書きます。
例では、ICounterServiceが呼ばれたらその実装であるCounterServiceを取得するようにしています。
また、scope=request_scopeによってインスタンスの生成をリクエストスコープに設定しています。

class AppModule(Module):
    def configure(self, binder: Binder):
        binder.bind(ICounterService, to=CounterService, scope=request_scope)
        return

FastAPIとinjectorを紐づけているのが以下の部分です。
先ほど設定したAppModuleでDIコンテナを作り、FastAPIアプリに対してミドルウェアを登録・関連付けします。

app = FastAPI()

injector = Injector([AppModule()])
app.add_middleware(InjectorMiddleware, injector=injector)
attach_injector(app=app, injector=injector)

実際にインジェクションされたインスタンスを取得する部分が以下です。
Injected()というヘルパー関数が用意されており、インターフェースとなる型を指定することでAppModuleに設定した実装が取得できます。

@app.get("/")
def get_root(counter_service: ICounterService = Injected(ICounterService)):
    before = counter_service.get_count()
    counter_service.increment()
    after = counter_service.get_count()

    return {
        "before": before,
        "after": after,
    }

DIとスコープの動作

基本的な使い方を紹介したので、実際にインジェクションされる例を記載します。
また、各スコープの種類(シングルトン、リクエストスコープ、トランジェント)を指定して、期待通りの動作になることも確認します。

※今更ですが、この記事で「トランジェント」というスコープの名称はASP.NET CoreのMicrosoft.Extensions.DependencyInjectionで使われている名称から引用しています。
DIにおいて「毎回インスタンスを生成する」ことを何と呼称すればよいのかはわかっておりません...。

まずはサンプルとなるクラスを用意します。
シングルトンなどの動作確認もしたいので、内部状態を持つようにしておきます。
(CounterServiceはインクリメントされたカウンター値を保持)

counter_service.py

from abc import ABC, abstractmethod


class ICounterService(ABC):
    @abstractmethod
    def increment(self) -> int:
        pass

    @abstractmethod
    def get_count(self) -> int:
        pass


class CounterService(ICounterService):
    def __init__(self):
        self._count = 0
        return

    def increment(self) -> int:
        self._count += 1
        return self._count

    def get_count(self) -> int:
        return self._count

ICounterServiceに依存するHogeServiceとFugaServiceも用意します。
コンストラクタにはinjectorの@injectデコレータを付けることで、型ヒントに合致する依存が自動的にインジェクションされるように設定できます。
do_something()メソッドを呼ぶと、ICounterServiceでカウンター値をインクリメントします。

hoge_service.py

from abc import ABC, abstractmethod
from injector import inject

from services.counter_service import ICounterService


class IHogeService(ABC):
    @abstractmethod
    def do_something(self) -> str:
        pass

    @abstractmethod
    def get_inner_count(self) -> int:
        pass


class HogeService(IHogeService):
    @inject
    def __init__(self, counter_service: ICounterService):
        self._counter_service = counter_service
        return

    def do_something(self) -> str:
        self._counter_service.increment()
        return "HogeService did something!"

    def get_inner_count(self) -> int:
        return self._counter_service.get_count()

fuga_service.py

from abc import ABC, abstractmethod
from injector import inject

from services.counter_service import ICounterService


class IFugaService(ABC):
    @abstractmethod
    def do_something(self) -> str:
        pass

    @abstractmethod
    def get_inner_count(self) -> int:
        pass


class FugaService(IFugaService):
    @inject
    def __init__(self, counter_service: ICounterService):
        self._counter_service = counter_service
        return

    def do_something(self) -> str:
        self._counter_service.increment()
        return "FugaService did something!"

    def get_inner_count(self) -> int:
        return self._counter_service.get_count()

APIの処理では、カウンター値のインクリメントや出力を行います。
リクエストを受けた時点、Hogeからインクリメントした時点、Fugaからインクリメントした時点でカウンター値を確認します。

@app.get("/")
def get_something(hoge_service: IHogeService = Injected(IHogeService), fuga_service=Injected(IFugaService)):
    print("=== Startup ===")
    print(f"count via hoge: {hoge_service.get_inner_count()}")
    print(f"count via fuga: {fuga_service.get_inner_count()}")

    hoge_service.do_something()

    print("=== After doing something by Hoge ===")
    print(f"count via hoge: {hoge_service.get_inner_count()}")
    print(f"count via fuga: {fuga_service.get_inner_count()}")

    fuga_service.do_something()

    print("=== After doing something by Fuga ===")
    print(f"count via hoge: {hoge_service.get_inner_count()}")
    print(f"count via fuga: {fuga_service.get_inner_count()}")

    return {
        "message": "This is injector endpoint"
    }

まずはCounterServiceをシングルトンにします。

from injector import Module, Binder, singleton


class AppModule(Module):
    def configure(self, binder: Binder):
        binder.bind(ICounterService, to=CounterService, scope=singleton)
        binder.bind(IHogeService, to=HogeService)
        binder.bind(IFugaService, to=FugaService)
        return

このAPIを2回呼び出したときの出力は以下です。
※視認性のため、リクエスト回数の表記や改行を加え、本題ではない内容は省略しています。

Hogeからインクリメントすると、Fuga側から取得した値もインクリメントされてます。
Fugaからインクリメントしても同様です。
また、リクエストを跨いでもカウンター値が保持されており、CounterServiceがシングルトンの挙動をしていることがわかります。

[リクエスト1回目]
=== Startup ===
count via hoge: 0
count via fuga: 0

=== After doing something by Hoge ===
count via hoge: 1
count via fuga: 1

=== After doing something by Fuga ===
count via hoge: 2
count via fuga: 2

[リクエスト2回目]
=== Startup ===
count via hoge: 2
count via fuga: 2

=== After doing something by Hoge ===
count via hoge: 3
count via fuga: 3

=== After doing something by Fuga ===
count via hoge: 4
count via fuga: 4

次に、トランジェントです。
トランジェント(=毎回インスタンスを生成する)はscopeをNone(または未指定)にします。

class AppModule(Module):
    def configure(self, binder: Binder):
        binder.bind(ICounterService, to=CounterService, scope=None)
        binder.bind(IHogeService, to=HogeService)
        binder.bind(IFugaService, to=FugaService)

        return

HogeのインクリメントはHogeのみ更新され、FugaのインクリメントはFugaのみ更新されます。
HogeとFugaが持つCounterServiceの実体は異なることがわかります。

[リクエスト1回目]
=== Startup ===
count via hoge: 0
count via fuga: 0

=== After doing something by Hoge ===
count via hoge: 1
count via fuga: 0

=== After doing something by Fuga ===
count via hoge: 1
count via fuga: 1

[リクエスト2回目]
=== Startup ===
count via hoge: 0
count via fuga: 0

=== After doing something by Hoge ===
count via hoge: 1
count via fuga: 0

=== After doing something by Fuga ===
count via hoge: 1
count via fuga: 1

最後に、リクエストスコープです。

from fastapi_injector import request_scope


class AppModule(Module):
    def configure(self, binder: Binder):
        binder.bind(ICounterService, to=CounterService, scope=request_scope)
        binder.bind(IHogeService, to=HogeService)
        binder.bind(IFugaService, to=FugaService)

        return

Hogeからインクリメントすると、Fuga側から取得した値もインクリメントされてます
Fugaからインクリメントしても同様です。
なので、HogeとFugaが持つCounterServiceの実体は同じであることがわかります。

しかし、リクエストを跨ぐとカウンター値が初期化されているので、異なるリクエストでは新たにCounterServiceが生成されていることがわかります。

[リクエスト1回目]
=== Startup ===
count via hoge: 0
count via fuga: 0

=== After doing something by Hoge ===
count via hoge: 1
count via fuga: 1

=== After doing something by Fuga ===
count via hoge: 2
count via fuga: 2

[リクエスト2回目]
=== Startup ===
count via hoge: 0
count via fuga: 0

=== After doing something by Hoge ===
count via hoge: 1
count via fuga: 1

=== After doing something by Fuga ===
count via hoge: 2
count via fuga: 2

DIコンテナを利用するメリット

FastAPIの標準では、DIコンテナを活用したときのような簡潔な記述ができません。
というのも、FastAPIの持つDI機構では以下のget_hoge_service()関数のような「実際にインスタンスを生成する処理」を指定する必要があります。
これはDIするクラスが増える度にこのような関数を用意する必要が出てきます。
(FugaServiceも欲しい場合は、「FugaServiceインスタンスを生成する処理」が必要)

from fastapi import Depends


def get_hoge_service() -> IHogeService:
    counter_service = CounterService()
    return HogeService(counter_service)


@app.get("/")
def get_something(hoge_service: IHogeService = Depends(get_hoge_service)):
    before = hoge_service.get_inner_count()
    hoge_service.do_something()
    after = hoge_service.get_inner_count()

    return {
        "before": before,
        "after": after,
    }

また、この例ではHogeService等からICounterServiceへの依存のみですが、
依存クラスが増えたり、あるいは、依存階層が増えていくとインスタンスの生成処理が煩雑になってしまうことが懸念されます。
injectorのDIコンテナ機能を利用することでインスタンス生成を自動化させ、DIの設定はAppModuleのような箇所にまとめて簡潔に書けるというのが個人的に便利だと思いました。

最後に

python-dependency-injectorとの比較ができていないですが、injector + fastapi-injectorでも十分便利そうなDIの仕組みが手に入ると思いました。
小規模なFastAPIの開発では標準のDIで十分かもしれませんが、規模によってはDIコンテナを使いたくなることもあるので、今回の記事が参考になると幸いです。