Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

第3章 Depends による依存性注入の実践

実務の Web API 開発では、認証・権限チェック、データベース接続の取得と破棄、共通設定の読み込みなど、多くのエンドポイントで共有される処理が発生します。

FastAPI は、これらを美しく解決する「依存性注入(Dependency Injection: DI)」の強力なシステムを備えています。本章では、Depends を活用した共通処理の設計手法を学びます。

3.1 依存性注入(Dependency Injection)とは何か

依存性注入とは、ある処理が必要とする機能やデータ(依存関係)を、その処理自身が直接生成するのではなく、外部から渡してもらう設計手法です。

例えば、ユーザー情報を取得するエンドポイントが自前で認証トークンの検証やデータベースの接続作成を行うと、コードが重複し、テストも困難になります。FastAPI の DI 機構を使うことで、共通処理を独立した関数として切り離し、エンドポイントの引数に注入することができます。

3.2 Depends の基本的な使い方

FastAPI では、fastapi.Depends を使用して依存関係を宣言します。

関数の引数に Depends(依存関数) を指定すると、FastAPI はエンドポイントが実行される直前にその依存関数を実行し、戻り値を引数に渡します。

from fastapi import Depends, FastAPI

app = FastAPI()


def pagination_params(skip: int = 0, limit: int = 10) -> dict[str, int]:
    return {"skip": skip, "limit": limit}


@app.get("/items")
def list_items(params: dict[str, int] = Depends(pagination_params)):
    return {
        "message": "Items fetched",
        "skip": params["skip"],
        "limit": params["limit"],
    }


@app.get("/users")
def list_users(params: dict[str, int] = Depends(pagination_params)):
    return {
        "message": "Users fetched",
        "skip": params["skip"],
        "limit": params["limit"],
    }

共通のクエリパラメータ処理(ページネーション情報)を 1 つの関数に集約でき、複数のエンドポイントで手軽に再利用できます。

3.3 認証・認可処理の共通化

DI の最も代表的な用途の 1 つが、API キーやトークンによる認証処理です。

HTTP ヘッダーから認証情報を取得し、不正な場合に即座に例外を発生させる依存関数を定義します。

from fastapi import Depends, FastAPI, HTTPException, Header, status

app = FastAPI()

VALID_API_KEYS = {"secret-token-12345": "AdminUser"}


def verify_api_key(x_api_key: str = Header(...)) -> str:
    user_name = VALID_API_KEYS.get(x_api_key)
    if not user_name:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or missing API Key",
        )
    return user_name


@app.get("/protected-data")
def get_protected_data(current_user: str = Depends(verify_api_key)):
    return {
        "status": "success",
        "data": "This is sensitive data",
        "operator": current_user,
    }

エンドポイント側は current_user: str = Depends(verify_api_key) と宣言するだけで、認証に合格したユーザー名のみを受け取ることができ、ビジネスロジックに集中できます。

3.4 データベースセッションのライフサイクル管理

リクエストごとにデータベース接続を開き、処理が終了したら確実にクローズする処理には、yield を用いたジェネレータ関数を利用します。

from collections.abc import Generator
from fastapi import Depends, FastAPI


class DatabaseSession:
    def __init__(self):
        print("DB接続をオープンしました")

    def close(self):
        print("DB接続をクローズしました")


def get_db() -> Generator[DatabaseSession, None, None]:
    db = DatabaseSession()
    try:
        yield db
    finally:
        db.close()


app = FastAPI()


@app.get("/records")
def list_records(db: DatabaseSession = Depends(get_db)):
    return {"message": "Records fetched using active DB session"}

yield の前にあるコードがエンドポイントの実行前に走り、レスポンスがクライアントへ返された後(または途中で例外が発生した場合でも)、finally ブロックのクリーンアップ処理が確実に実行されます。

3.5 テスト時の依存性オーバーライド

DI 機構を採用する大きな利点は、単体テストや結合テストが極めて容易になる点です。

FastAPI の app.dependency_overrides 辞書を使うと、テスト実行時のみ本物の依存関係をモック(ダミー)関数に差し替えることができます。

from fastapi.testclient import TestClient

client = TestClient(app)


def mock_verify_api_key():
    return "TestMockUser"


# テスト用に依存関係を差し替え
app.dependency_overrides[verify_api_key] = mock_verify_api_key

response = client.get("/protected-data")
assert response.status_code == 200
assert response.json()["operator"] == "TestMockUser"

# テスト終了後に元に戻す
app.dependency_overrides.clear()

本物の認証サーバーや外部データベースを用意することなく、確実で高速なテストを実行できます。