ホーム

依存性注入

from typing import Annotated,from fastapi import Dependの具体的な手法

依存性注入(Dependency Injection)in FastAPI

FastAPI では強力かつ直感的な依存性注入システムが提供されています。特にtypingモジュールのAnnotatedfastapiDependsを組み合わせることで、効率的で保守性の高いコードを作成できます。この報告書では、これらの技術を詳しく解説します。

依存性注入とは

依存性注入とは、プログラミングにおいて、コード(この場合はパス操作関数)が動作するために必要なもの(「依存関係」)を宣言する方法です。システム(FastAPI)が必要な依存関係をコードに提供(「注入」)します[^1]。

これは以下のようなケースで特に有用です:

  • ロジックの共有
  • データベース接続の共有
  • セキュリティ、認証、ロールの要件の強制
  • コードの重複を最小限に抑えながら様々な機能を実現[^1]

基本的な依存性注入とDepends

シンプルな例

FastAPI での最も基本的な依存性注入は、Depends関数を使用します:

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

この例では、common_parameters関数が依存関係として定義され、2 つの異なるエンドポイントで再利用されています

Annotatedを使用した依存性注入

Python 3.9 から導入されたAnnotatedは、型ヒントにメタデータを追加する機能を提供します。FastAPI では、これを依存性注入と組み合わせることで、より明示的で読みやすいコードを書くことができます

Annotatedとは

Annotatedは、Python の型ヒントに追加のメタデータを付与するための機能です。これにより、型情報だけでなく、その型に関連する追加情報も表現できるようになります

AnnotatedDependsの組み合わせ

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

この方法では、commonsパラメータがdict型であり、さらにcommon_parameters関数の戻り値を受け取るという情報が明示的に示されています

Annotatedの利点

  1. 明示的な型情報: 引数の型と依存関係が明確に分離されます
  2. コードの可読性: 型とその操作が分かりやすく記述できます
  3. 再利用性の向上: 依存関係の定義を変数に代入して再利用できます
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

この例では、CommonsDepという型エイリアスを定義して再利用しています

高度な依存性注入パターン

サブ依存関係

FastAPI では、依存関係がさらに別の依存関係(サブ依存関係)を持つことができます。これにより、依存関係を必要な深さまで階層化できます

from typing import Annotated
from fastapi import Cookie, Depends, FastAPI

app = FastAPI()

def query_extractor(q: str | None = None):
    return q

def query_or_cookie_extractor(
    q: Annotated[str, Depends(query_extractor)],
    last_query: Annotated[str | None, Cookie()] = None,
):
    if not q:
        return last_query
    return q

@app.get("/items/")
async def read_query(
    query_or_default: Annotated[str, Depends(query_or_cookie_extractor)],
):
    return {"q_or_cookie": query_or_default}

この例では、query_or_cookie_extractorquery_extractorに依存しています。FastAPI は自動的にこれらの依存関係を解決します

クラスを使用した依存性注入

関数だけでなく、Python クラスも依存関係として使用できます

from typing import Annotated
from fastapi import Depends, FastAPI

app = FastAPI()

class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends()]):
    return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

この方法の利点は、単なる辞書ではなくクラスのインスタンスを取得できるため、IDE の補完機能などが使いやすくなることです[^11]。

yieldを使用したリソース管理

依存関係関数でyieldを使用すると、リクエスト処理後にクリーンアップ処理を実行できます。これはデータベース接続やファイルハンドラなどのリソース管理に特に有用です

from fastapi import FastAPI, Depends
from contextlib import contextmanager

app = FastAPI()

@contextmanager
def get_db():
    db = "Database Connection"
    yield db  # この値が呼び出し元に返される
    # クリーンアップコードはここに
    print("データベース接続を閉じています")

@app.get("/items/")
def read_items(db: str = Depends(get_db)):
    return {"db": db}

依存性注入の応用例

セキュリティと認証

依存性注入は、認証や認可の実装に特に有効です

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()

def get_current_user():
    # ユーザー取得ロジック
    return user

def get_admin_user(current_user = Depends(get_current_user)):
    if not current_user.is_admin:
        raise HTTPException(status_code=403, detail="権限が不足しています")
    return current_user

@app.get("/admin")
def read_admin_data(admin_user = Depends(get_admin_user)):
    return {"message": "管理者エリアへようこそ!"}

パス操作デコレータでの依存性注入

パス操作関数の引数だけでなく、デコレータにも依存関係を追加できます

from fastapi import Depends, FastAPI

app = FastAPI()

def verify_token(x_token: str):
    # トークン検証ロジック
    pass

@app.get("/items/", dependencies=[Depends(verify_token)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

スレッドセーフなカウンター管理を実現するクラスです。

# threadSafecounter.py
from threading import Lock

class ThreadSafeCounter:
    def __init__(self):
        self._count = 0
        self._lock = Lock()  # 排他制御用ロック

    def increment(self):
        with self._lock:  # ロック取得
            self._count += 1
            return self._count

    def set(self, num: int):
        with self._lock:  # ロック取得
            self._count = num
            return self._count

    def get(self):
        return self._count

を利用するシーンを考えます。

from fastapi import FastAPI, Depends
from typing import Annotated
from threadSafecounter import ThreadSafeCounter

counter_instance = ThreadSafeCounter()
countobjDep = Annotated[ThreadSafeCounter, Depends(lambda:counter_instance)]

@app.get("/set_useobj/{num}")
async def set_count(num: int, cntobj:countobjDep):
    cntobj.set(num)
    return {"count": cntobj.get()}

def set_counter(num: int):
    return counter_instance.set(num)

setcountDep = Annotated[int, Depends(set_counter)]
@app.get("/set/{num}")
async def set_count(num: int, cnt:setcountDep):
    return {"count": cnt}
  • _lock で排他制御し、マルチスレッド環境でも安全に値を更新
  • increment(): 現在値 +1 して返す
  • set(num): 任意の数値に設定して返す

2️⃣ スコープの考慮

  • counter_instance はグローバル変数として生成
  • マルチプロセス環境では別インスタンスが生成される可能性あり

実装のポイントは、Lock() による排他制御で値の整合性を保証している点です。FastAPI の非同期処理環境でも安全に動作します。