Dependency Injection System — The Core Design of FastAPI
依存性注入は LEGO ブロックの接口のようなものです。各ブロックは自分が何を提供すべきかだけを気にし, 誰が使っているかを知る必要はありません。データベース接続, 現在のユーザー, 権限チェックはネストされたレイヤーを通じて自動的に解決されます。
1. 学ぶ内容
Depends()の基本:関数依存, クラス依存, ネストされたサブ依存- 依存のライフサイクル:リクエストレベル vs アプリケーションレベル (
yield依存とリソースクリーンアップ) - グローバル依存とルーティンググループ依存:
dependencies=[Depends(...)] - 依存のオーバーライドとテスト:
app.dependency_overridesの実践的ティップス - Alice シナリオ:PriceTracker の DB セッション依存, 現在のユーザー依存, 権限チェック依存 — 3層ネスト DI チェーン
2. Alice のリアルストーリー
(1) ペインポイント:各エンドポイントでデータベースとユーザー情報の取得を繰り返す
Alice の PriceTracker には20のエンドポイントがあり, それぞれでデータベース接続を開き, JWT トークンを検証し, 現在のユーザーを取得し, 権限をチェックする必要があります。彼女は各エンドポイント関数に同じ15行の初期化コードをコピーアンドペーストしているため, データベース接続方法が変更された場合, 20箇所を個別に修正する必要があります。
(2) 依存性注入の解決策
FastAPI の依存性注入は, 繰り返しのロジックを再利用可能な依存関数に抽出します。エンドポイントは db = Depends(get_db) と宣言するだけで, FastAPI が自動的に解決・注入し, サブ依存の再帰的解決も含みます。
PYTHON
from fastapi import Depends
async def get_db():
db = Database()
yield db
db.close()
async def get_current_user(token: str, db=Depends(get_db)):
return verify_token(token, db)
@app.get("/products")
async def list_products(user=Depends(get_current_user), db=Depends(get_db)):
# user と db は自動注入される
...
(3) 成果
データベース接続, ユーザー認証, 権限チェック — 以前は20のエンドポイントそれぞれに15行のコードが必要だったものが, 3つの依存関数に統合され, 1箇所の変更がグローバルに反映されるようになりました。テスト時は app.dependency_overrides[get_db] = lambda: mock_db と1行書くだけで, 全エンドポイントのデータベース情報を置き換えられます。
3. Depends() の基本
(1) 関数依存
(1) ▶ サンプル:シンプルな関数依存
PYTHON
from fastapi import FastAPI, Depends
app = FastAPI()
def common_parameters(
q: str | None = None,
skip: int = 0,
limit: int = 100,
):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/products")
async def list_products(commons: dict = Depends(common_parameters)):
return commons
@app.get("/prices")
async def list_prices(commons: dict = Depends(common_parameters)):
return commons
出力:
TEXT
# 関数定義成功
(2) ▶ サンプル:クラス依存
PYTHON
from fastapi import FastAPI, Depends, Query
class CommonQueryParams:
def __init__(
self,
q: str | None = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=200),
):
self.q = q
self.skip = skip
self.limit = limit
app = FastAPI()
@app.get("/products")
async def list_products(commons: CommonQueryParams = Depends(CommonQueryParams)):
return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}
出力:
TEXT
# 関数定義成功
(2) 関数依存 vs クラス依存
| 側面 | 関数依存 | クラス依存 |
|---|---|---|
| 定義方法 | def get_xxx() |
class Xxx: + __init__ |
| ユースケース | シンプルなロジック, 単一呼び出し | 状態が必要, 複数メソッド |
| パラメータ解析 | 関数パラメータの自動解析 | __init__ パラメータの自動解析 |
| 再利用性 | 高い | 中程度 |
| 推奨度 | 第一選択 | 複雑なシナリオで使用 |
4. ネストされたサブ依存
(1) 依存解決ツリー
graph TD
A[get_current_user] --> B[get_db]
A --> C[get_token_from_header]
C --> D[OAuth2PasswordBearer]
E[require_admin] --> A
E --> F[check_subscription]
style A fill:#e1f5fe
style E fill:#fff3e0
FastAPI は依存ツリーを再帰的に解析します:require_admin → get_current_user → get_db。各依存は1回だけインスタンス化されます (同じリクエスト内でキャッシュ)。
(1) ▶ サンプル:ネストされたサブ依存 — DB → User
PYTHON
from fastapi import FastAPI, Depends, HTTPException, Header
app = FastAPI()
# レベル1:データベースセッション
async def get_db():
db = {"connection": "active"}
try:
yield db
finally:
db["connection"] = "closed"
# レベル2:現在のユーザーを取得 (DB に依存)
async def get_current_user(
authorization: str = Header(...),
db: dict = Depends(get_db),
):
if not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Invalid token")
token = authorization.replace("Bearer ", "")
# 簡略化:本番では JWT を検証
user = {"id": 1, "username": "alice", "role": "admin"}
return user
# レベル3:管理者権限が必要 (現在のユーザーに依存)
async def require_admin(user: dict = Depends(get_current_user)):
if user["role"] != "admin":
raise HTTPException(status_code=403, detail="Admin required")
return user
@app.get("/admin/stats")
async def admin_stats(admin: dict = Depends(require_admin)):
return {"admin": admin["username"], "total_products": 1000000}
出力:
TEXT
# 関数定義成功
5. 依存のライフサイクルと yield
(1) リクエストレベル vs アプリケーションレベル
| ライフサイクル | 宣言方法 | スコープ | 典型的な用途 |
|---|---|---|---|
| リクエストレベル | def dep(): yield x; cleanup |
リクエストごとに作成・破棄 | DB セッション, Redis 接続 |
| アプリケーションレベル | @app.on_event("startup") |
アプリケーション起動時に作成 | エンジン初期化, 接続プール |
(1) ▶ サンプル:yield 依存とリソースクリーンアップ
PYTHON
from fastapi import FastAPI, Depends
app = FastAPI()
# リクエストスコープの依存 (クリーンアップ付き)
async def get_db_session():
# セットアップ:セッションを作成
session = {"id": "session-123", "active": True}
print(f"DB session opened: {session['id']}")
try:
yield session # これがエンドポイントに注入される
finally:
# クリーンアップ:セッションを閉じる (レスポンス後に実行)
session["active"] = False
print(f"DB session closed: {session['id']}")
@app.get("/products")
async def list_products(db: dict = Depends(get_db_session)):
print(f"Using DB session: {db['id']}")
return {"session_id": db["id"]}
出力 (サーバーログ):
TEXT
DB session opened: session-123
Using DB session: session-123
DB session closed: session-123
6. グローバル依存とルーティンググループ依存
(1) 異なるスコープの依存宣言
(1) ▶ サンプル:ルーティンググループ依存
PYTHON
from fastapi import FastAPI, Depends, APIRouter, Header, HTTPException
async def verify_api_key(x_api_key: str = Header(...)):
if x_api_key != "secret-key-123":
raise HTTPException(status_code=401, detail="Invalid API key")
return x_api_key
app = FastAPI()
# 公開ルート - 認証不要
public_router = APIRouter()
@public_router.get("/health")
async def health():
return {"status": "healthy"}
# 保護ルート - このグループの全ルートに API キーが必要
protected_router = APIRouter(dependencies=[Depends(verify_api_key)])
@protected_router.get("/products")
async def list_products():
return [{"id": 1, "name": "Widget"}]
@protected_router.post("/products")
async def create_product():
return {"id": 2, "name": "New Product"}
# ルーターの登録
app.include_router(public_router)
app.include_router(protected_router, prefix="/api/v1")
出力:
TEXT
# 関数定義成功
| 依存スコープ | 宣言場所 | 影響範囲 |
|---|---|---|
| エンドポイントレベル | @app.get("/", dependencies=[...]) |
単一エンドポイント |
| ルーティンググループレベル | APIRouter(dependencies=[...]) |
ルーティンググループ内の全エンドポイント |
| グローバル | FastAPI(dependencies=[...]) |
アプリケーション内の全エンドポイント |
7. 依存のオーバーライドとテスト
(1) dependency_overrides
テスト時には, 本番用の依存 (データベース接続や外部 API など)を置き換える必要があります。app.dependency_overrides を使用すると, ソースコードを変更せずに依存の実装を置き換えられます。
(1) ▶ サンプル:テストでデータベース依存を置き換え
PYTHON
from fastapi.testclient import TestClient
from fastapi import FastAPI, Depends
app = FastAPI()
# 本番用の依存
async def get_db():
return {"type": "postgresql", "host": "prod-db"}
# テスト用のモック依存
def get_mock_db():
return {"type": "sqlite", "host": "memory"}
@app.get("/db-info")
async def db_info(db: dict = Depends(get_db)):
return db
# テスト内:
def test_db_info():
app.dependency_overrides[get_db] = get_mock_db
client = TestClient(app)
response = client.get("/db-info")
assert response.json() == {"type": "sqlite", "host": "memory"}
# クリーンアップ
app.dependency_overrides.clear()
出力:
TEXT
# 関数定義成功
❓ よくある質問
Q Depends 依存は複数回呼び出されますか?
A 同じリクエスト内では, 各依存は1回だけ実行され, 結果はキャッシュされて再利用されます。異なるリクエスト間では, 依存は再実行されます。
Q
yield 依存のクリーンアップ関数はいつ実行されますか?A レスポンスがクライアントに送信された後に実行されます。エンドポイントが例外をスローした場合でも, クリーンアップ関数は実行されます (
try/finally と同様)。Q 依存はどれくらい深くネストできますか?
A 厳密な制限はありませんが, 3レベルを超えるとデバッグが困難になります。PriceTracker の3層 DI チェーン (get_db → get_current_user → require_admin)が推奨される上限です。
Q 同期と非同期の依存は一緒に使えますか?
A はい。FastAPI は同期・非同期の依存を自動的に処理します。同期の依存はスレッドプールで実行され, 非同期の依存はイベントループで実行されます。
Q
dependency_overrides は他のテストに影響しますか?A はい,
app オブジェクトを変更するため影響します。テスト終了後に app.dependency_overrides.clear() を呼び出すか, フィクスチャで管理してください。Q クラス依存の init 引数はどのように注入されますか?
A 関数依存と同様に, init
📖 まとめ
Depends()は繰り返しのロジックを再利用可能なヘルパー関数やクラスに抽出し, エンドポイントは宣言するだけで実装を気にする必要はありません- サブ依存の自動的な再帰的解決により, 同じリクエスト内でキャッシュして重複実行を回避します
yieldは依存のリクエストレベルのライフサイクル管理を実現し, リソースの作成とクリーンアップのペアを保証します- ルーティンググループ依存 (
APIRouter(dependencies=[...]))により, 一群のエンドポイントが認証とバリデーションロジックを共有できます app.dependency_overridesは強力なテストツールで, 1行のコードで全エンドポイントの実際の依存を置き換えられます
📝 練習問題
- 基本問題 (難易度 ⭐):データベース接続をシミュレートする辞書を返すヘルパー関数
get_dbを作成し, 2つのエンドポイントに注入して使用してください。ヒント:db: dict = Depends(get_db) - 応用問題 (難易度 ⭐⭐):2層のネストされた依存を実装してください:
get_db→get_current_user(リクエストから Authorization ヘッダーを読み取る)。/meエンドポイントに現在のユーザーの情報を注入してください。ヒント:user: dict = Depends(get_current_user) - チャレンジ問題 (難易度 ⭐⭐⭐):PriceTracker の3層 DI チェーンを実装してください:
get_db(yield + クリーンアップ)→get_current_user(トークン検証)→require_subscription("pro")(サブスクリプションレベル確認)。テスト時にdependency_overridesで DB 依存を置き換えてください。ヒント:yield依存 +app.dependency_overrides[get_db] = mock_fn
---|
