🎴
FastAPI の Dependency Injection (Depends + Injector )

Yuto Moriyasu
2025/01/11に公開
はじめまして、株式会社 neoAI というところでソフトウェアエンジニアをしている Moriyasu といいます。

今回は弊社として初めてのソフトウェアのブログを担当させていただくことになりました。
さて、年末年始はみなさんいかがお過ごしでしたか？

僕は暇だったので、自社プロダクトを Flask から FastAPI に移行するコーディングをずっとやっていました。FastAPI を"完全に理解"したので、今回は FastAPI での Dependency Injection (依存性の注入) について解説します。

 前提今回出てくる技術について先にさらっておきます。

FastAPI: 高速でモダンな API を構築するための Python フレームワーク

SQLAlchemy: Python の ORM ライブラリ

Injector: Python の Dependency Injection ライブラリ
本題に入る前に、FastAPI の Dependencies と Middleware、Injector ライブラリについて軽く説明します。

 DependenciesFastAPI には元々 Dependency Injection の仕組みが搭載されており、Depends を使うことで Path Operation に依存関係を注入することができます。
https://fastapi.tiangolo.com/tutorial/dependencies/
以下のように関数を定義して、Depends を使って Path Operation の引数に渡すと、不思議なことに get_message が実行され、その返り値が message という引数に代入されます。
from fastapi import FastAPI, Depends

app = FastAPI()

def get_message():
    return "Hello, world!"

@app.get("/hello")
async def hello(message: str = Depends(get_message)):
    return {"message": message}
Depends は Path Operation だけでなく、APIRouter や FastAPI でも利用できます。

Depends に入れる関数にはいろんな引数が定義でき、Request インスタンスや Depends を入れることもできます。
Path Operation の関数で定義できる引数は全部いけるって書いてました。
A FastAPI dependency function can take any of the arguments that a normal endpoint function can take.
https://stackoverflow.com/questions/68668417/is-it-possible-to-pass-path-arguments-into-fastapi-dependency-functions
Request インスタンスは Path Operation で利用できるから、Depends の関数に入れられるものだと思っていますが、本当かどうかは分かりません。
By declaring a path operation function parameter with the type being the Request FastAPI will know to pass the Request in that parameter.
https://fastapi.tiangolo.com/advanced/using-request-directly/#use-the-request-object-directly

 MiddlewareFastAPI の Middleware はリクエストやレスポンスの前後に好きな処理を追加できます。
https://fastapi.tiangolo.com/tutorial/middleware/
BaseHTTPMiddleware を継承したクラスの dispatch メソッドを実装するだけでOKです。FastAPI のインスタンスに add_middleware を使って登録できます。
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from typing import Callable

class MyMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, some_attribute: str):
        super().__init__(app)
        self.some_attribute = some_attribute

    async def dispatch(self, request: Request, call_next: Callable):
        print(f"リクエスト前処理: {self.some_attribute}")

        response = await call_next(request)

        print(f"レスポンス後処理: {self.some_attribute}")

        return response


app = FastAPI()

app.add_middleware(MyMiddleware, some_attribute="test")
デコレータもあるので、短い処理はこれで十分ですね。
from fastapi import FastAPI, Request
from typing import Callable

app = FastAPI()

@app.middleware("http")
async def my_middleware(request: Request, call_next: Callable):
    # リクエスト処理の前に行う処理
    print("リクエスト前処理")

    response = await call_next(request)

    # レスポンス処理の後に行う処理
    print("レスポンス後処理")

    return response

 InjectorInjector は Python で Dependency Injection を実装するためのライブラリです。

Python には他にも Dependency Injector というライブラリがあるのですが、Injector の方がシンプルに書けて簡単だったので採用しました。

いろんな使い方がありますが、僕は Module クラスを使ってシンプルに実装しています。
# リポジトリの定義
from abc import ABC, abstractmethod

class IUserRepository(ABC):
    @abstractmethod
    def find_by_tenant_id(self, tenant_id: TenantId) -> list[User]:
        pass

# リポジトリの実装
class UserRepository(IUserRepository):
    def find_by_tenant_id(self, tenant_id: TenantId) -> list[User]:
        # 実装

# Module の定義
from injector import Module, provider

class DatabaseModule(Module):
    @provider
    def user_repo(self) -> IUserRepository:
        return UserRepository()

# ユースケースの定義
from abc import ABC, abstractmethod

class IUserUseCase(ABC):
    @abstractmethod
    def get_users_by_tenant_id(self, tenant_id: TenantId) -> list[User]:
        pass

# ユースケースの実装
class UserUseCase(IUserUseCase):
    @inject
    def __init__(
        self,
        user_repo: IUserRepository,
    ) -> None:
        self.user_repo = user_repo

    def get_users_by_tenant_id(self, tenant_id: TenantId) -> list[User]:
        return self.user_repo.find_by_tenant_id(tenant_id)

# Injector のインスタンス作成
from injector import Injector

injector = Injector([DatabaseModule()])

# ユースケースの実体を取得
user_interactor = injector.get(UserUseCase)
user_interactor.get_users_by_tenant_id(TenantId(1))

 本題では、FastAPI での実装を解説します。ORM には SQLAlchemy を使用しています。

SQLAlchemy のセッションをどうやってリポジトリに注入するかがポイントなので、注意して見てみてください。

※ import 文は省略してるので、同じクラス名を追ってください

 SQLAlchemy Session の作成まずは、SQLAlchemy の engine と session の設定を実装します。

create_engine で SQLAlchemy のエンジンを作成し、sessionmaker のインスタンスである SessionFactory を呼び出すことでセッションを作成できます。
import os

from fastapi import Request
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

# データベースの設定値
db_host = os.environ.get("DB_HOST", "")
db_name = os.environ.get("DB_NAME", "")
db_user = os.environ.get("DB_USER", "")
db_password = os.environ.get("DB_PASSWORD", "")

# engine の作成
engine = create_engine(f"postgresql://{db_user}:{db_password}@{db_host}/{db_name}?client_encoding=utf8")

# sessionmaker の作成
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def set_db_session_to_request(request: Request, session: Session):
    request.state.__setattr__("session", session)

def get_db_session_from_request(request: Request) -> Session:
    session = getattr(request.state, "session", None)
    if session is None:
        raise ValueError("Session is not set to request")
    if not isinstance(session, Session):
        raise ValueError("Session is not an instance of Session")
    return session
また、request.state にセッションをセットする関数 set_db_session_to_request と、request.state からセッションを取得する関数 get_db_session_from_request を定義しています。

FastAPI の Request.state にはリクエスト中に使用できるオブジェクトを保存しておくことができます。Starlette の方のドキュメントの隅っこに書いてあります。
If you want to store additional information on the request you can do so using request.state.
https://www.starlette.io/requests/#other-state
ミドルウェアでリクエストごとにセッションを作成し、request.state にセットします。

with 文を利用することで、リクエストの終了時にセッションを閉じることができます。
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def set_db_session(request: Request, call_next):
    with SessionFactory() as session:
        set_db_session_to_request(request, session)
        response = await call_next(request)
        return response

 リポジトリの実装リポジトリは、sqlalchemy.orm.Session をコンストラクタで受け取る形にしています。これにより、各メソッドからリクエストで共通なセッションを使用することができます。
from sqlalchemy import select
from sqlalchemy.orm import Session

class UserRepository(IUserRepository):
    def __init__(self, session: Session):
        self.session = session

    def find_by_tenant_id(self, tenant_id: TenantId) -> list[User]:
        users = (
            self.session.execute(
                select(UserModel)
                .where(UserModel.tenant_id == tenant_id.value)
            )
            .unique()
            .scalars()
            .all()
        )
        return [user.to_domain() for user in users]

 Module の実装Module クラスを使用して、リポジトリを登録します。DatabaseModule は、Session をコンストラクタで受け取り、tenant_repo と user_repo というプロバイダメソッドを通じて、それぞれのインターフェースに対応するリポジトリの実装を返します。
from injector import Module, provider
from sqlalchemy.orm import Session

class DatabaseModule(Module):
    def __init__(self, session: Session) -> None:
        self.session = session

    @provider
    def tenant_repo(self) -> ITenantRepository:
        return TenantRepository(self.session)

    @provider
    def user_repo(self) -> IUserRepository:
        return UserRepository(self.session)

 DI の実装Injector のインスタンスを返す関数 get_injector に、Depends を使用してセッションを注入します。先ほど作成した get_db_session_from_request は Request のインスタンスを引数に受け取る関数で、Depends の中で使用することができます。

これで、Injector が作成される際に、データベースセッションが自動的に渡されるようになります。
from fastapi import Depends
from injector import Injector
from sqlalchemy.orm import Session

def get_injector(
    session: Session = Depends(get_db_session_from_request),  # noqa: B008
) -> Injector:
    return Injector(
        [
            DatabaseModule(session),
        ]
    )

 Path Operation の実装get_user_interactor を使用し、Injector のインスタンスからユースケースの実体を取り出して、Route の関数に渡します。Depends を繰り返し使用することで、必要な依存関係が連鎖的に解決されるようになっています。
from fastapi import APIRouter, Body, Depends, Request
from injector import Injector

def get_user_interactor(
    injector: Injector = Depends(get_injector),  # noqa: B008
) -> IUserUseCase:
    return injector.get(UserUseCase)

user_router = APIRouter()

@user_router.post("/users")
def create_user(
    request: Request,
    user_interactor: IUserUseCase = Depends(get_user_interactor),  # noqa: B008
    param: CreateUserParam = Body(...),  # noqa: B008
):
    tenant_id = get_tenant_from_request(request).id
    return user_interactor.get_users_by_tenant_id(tenant_id)
get_db_session_from_request → get_injector → get_user_interactor の順に関数が実行され、ユースケースの実体を Path Operation に渡すことができます。
Depends のこのような使用方法は Sub-dependencies に書いてあるので読んでみてください。

https://fastapi.tiangolo.com/tutorial/dependencies/sub-dependencies/
一方でセッションは、Middleware → Request.state → Injector → Module → Repository の順で渡され、各リクエストで使用できるようになります。
ちなみに、セッションを Request.state にセットせずに、get_db_session_from_request をセッションを返す関数に置き換えてもいいです。

最初はその方法で行こうと思っていたのですが、認証のミドルウェア (Depends が使えなかった) の中でも同じセッションを使いたいので、Request インスタンスにセッションを保持しておく形にしました。そういった要件がない場合は直接セッションを作成して注入する実装でも良さそうですね。

 最後にRequest.state と Depends を利用して、かなりスッキリ書けましたね。

Sub-dependencies という概念を知ったときは脳汁が出ました。ぜひ活用してみてください。
Flask から FastAPI への移行についてはまだまだ紹介できることがたくさんあるので、シリーズ化できればなーと思っています。次は認証認可について書きます。また、よろしくです。
neoAIPublication
neoAIは生成AIに特化した東大松尾研発スタートアップです。
Discussion

ログインするとコメントできます