依存性注入
from typing import Annotated,from fastapi import Dependの具体的な手法
依存性注入(Dependency Injection)in FastAPI
FastAPI では強力かつ直感的な依存性注入システムが提供されています。特にtyping
モジュールのAnnotated
とfastapi
のDepends
を組み合わせることで、効率的で保守性の高いコードを作成できます。この報告書では、これらの技術を詳しく解説します。
依存性注入とは
依存性注入とは、プログラミングにおいて、コード(この場合はパス操作関数)が動作するために必要なもの(「依存関係」)を宣言する方法です。システム(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 の型ヒントに追加のメタデータを付与するための機能です。これにより、型情報だけでなく、その型に関連する追加情報も表現できるようになります
Annotated
とDepends
の組み合わせ
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
の利点
- 明示的な型情報: 引数の型と依存関係が明確に分離されます
- コードの可読性: 型とその操作が分かりやすく記述できます
- 再利用性の向上: 依存関係の定義を変数に代入して再利用できます
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_extractor
がquery_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 の非同期処理環境でも安全に動作します。