FastAPI Depends の使い方 – 認証・DB接続を効率的に実装する

目次

FastAPI の Depends で依存注入を使いこなそう

依存注入(Dependency Injection)とは

FastAPI を使っていて、複数のエンドポイントで同じ認証チェックを繰り返し書いている、データベース接続を毎回手動で開閉している、という経験はありませんか?

今回は、FastAPI の Depends

もしご自身で先に公式ドキュメントをご覧になられる方は下のリンクから眺めてみてください。
Dependencies – FastAPI ー 公式ドキュメント(英語)

対象となる方

  • FastAPI の基本的な使い方は知っている方
  • 複数のエンドポイントで同じ処理を繰り返している方
  • テストしやすいコードを書きたい方

 ※ このドキュメントは FastAPI 0.141.1 以上で書いていきます。

Depends なし vs Depends ありの比較

Depends なし(悪い例)

認証チェックがエンドポイントごとに重複してしまっています。

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/api/documents")
async def get_documents():
    token = request.headers.get("Authorization")
    if not token:
        raise HTTPException(status_code=401, detail="認証が必要です")
    user_id = await verify_token(token)
    return {"documents": [...]}

@app.post("/api/documents")
async def create_document():
    token = request.headers.get("Authorization")
    if not token:
        raise HTTPException(status_code=401, detail="認証が必要です")
    user_id = await verify_token(token)
    return {"success": True}

認証チェックコードが 3 回繰り返されていますね。エンドポイントが 20 個、50 個となるにつれ、管理が大変になります。

Depends あり(良い例)

from fastapi import FastAPI, Depends, HTTPException
from dataclasses import dataclass

app = FastAPI()

@dataclass
class CurrentUser:
    user_id: int
    email: str

async def get_current_user(token: str = Header(...)) -> CurrentUser:
    if not token:
        raise HTTPException(status_code=401)
    user_id = await verify_token(token)
    return CurrentUser(user_id=user_id, email="user@example.com")

@app.get("/api/documents")
async def get_documents(user: CurrentUser = Depends(get_current_user)):
    return {"documents": [...], "user_id": user.user_id}

@app.post("/api/documents")
async def create_document(user: CurrentUser = Depends(get_current_user)):
    return {"success": True, "user_id": user.user_id}

認証ロジックが get_current_user() 1 箇所に集中しました。全エンドポイントは Depends(get_current_user) の 1 行だけで認証チェックが完了します。

Depends の基本

FastAPI の Depends

async def get_current_user() -> CurrentUser:
    return CurrentUser(user_id=1, email="user@example.com")

@app.get("/api/me")
async def get_me(user: CurrentUser = Depends(get_current_user)):
    return {"user_id": user.user_id, "email": user.email}

流れ:

  1. リクエスト受信
  2. FastAPI が get_current_user() を実行
  3. 戻り値(CurrentUser)を生成
  4. エンドポイントハンドラに user として渡す
  5. レスポンス送信

実装例:複数の依存関数を組み合わせる

では、より実践的な例を見てみましょう。認証チェック+データベースセッション管理を組み合わせます。

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncSession:
    async with async_session_maker() as session:
        yield session
@app.get("/api/users/{user_id}/documents")
async def get_user_documents(
    user_id: int,
    user: CurrentUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    from sqlalchemy import select
    from models import Document

    result = await db.execute(
        select(Document).where(Document.user_id == user_id)
    )
    documents = result.scalars().all()
    return {"documents": documents, "requested_by": user.user_id}

3 つの依存関数(get_current_userget_db)が自動的に処理されます。エンドポイントハンドラは「ビジネスロジックだけに専念」できます。

テスタビリティの向上

依存注入の大きなメリットは、テストが書きやすくなることです。

import pytest
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_get_user_documents():
    mock_user = CurrentUser(user_id=1, email="test@example.com")
    mock_db = AsyncMock()

    mock_result = AsyncMock()
    mock_result.scalars.return_value.all.return_value = [
        {"id": 1, "title": "Document 1"},
        {"id": 2, "title": "Document 2"},
    ]
    mock_db.execute.return_value = mock_result

    result = await get_user_documents(
        user_id=1,
        user=mock_user,
        db=mock_db,
    )

    assert result["requested_by"] == 1
    assert len(result["documents"]) == 2

Depends を使えば、モック依存関数を直接注入できるため、テストが簡潔になります。

保守性の比較

「認証ロジックを JWT から Cognito に変更する」という場面を想定してみましょう。

Depends なし(最悪)

50 個のエンドポイントすべてで認証チェックコードを変更する必要があります。

Depends あり(簡単)

async def get_current_user(token: str = Header(...)) -> CurrentUser:
    # 変更前:
    # user_id = await verify_jwt_token(token)

    # 変更後:
    user_id = await verify_cognito_token(token)

    return CurrentUser(user_id=user_id, email="user@example.com")

# 全エンドポイントは変更不要
# Depends(get_current_user) のままで OK

まとめ

以上で FastAPI の Depends(依存注入)について説明を終えたいと思います。

Depends は「認証」「DB セッション」「ロギング」など、複数エンドポイントで共有される処理を効率的に実装するための、FastAPI のキラー機能です。

「同じコードを何度も書いている」と感じたら、それは Depends で依存関数に切り出すチャンスかもしれませんね。DRY(Don’t Repeat Yourself)原則に従うことで、保守性とテスタビリティが大きく向上します。

依存注入のスタイルを身につけることで、FastAPI での開発がぐっと効率化してきますね ^^v

ということで、ではまた〜

よかったらシェアしてね!
  • URLをコピーしました!
  • URLをコピーしました!

この記事を書いた人

Web Developer / Educator