woman wearings coop neck floral top using her apple brand macbook
Photo by JÉSHOOTS on Pexels.com
目次

スケールする設計へ:FastAPIのAPIRouterとDependsで実現するモジュール分割・依存性注入・設定管理


✅ まずは要約(インバーテッドピラミッド)

  • この記事でできること
    FastAPIの核である APIRouter(モジュール分割)Depends(依存性注入) を使って、保守しやすく拡張に強いAPI構成を組み立てます。DBセッションや認証、設定値の受け渡しを「依存」として扱い、ルーター単位の共通依存エンドポイント単位の依存テスト時の依存差し替えまで、実務で困らないパターンに落とし込みます。
  • 主なトピック
    1. APIRouterでの機能ごとの分割と include_router のベストプラクティス
    2. DependsAnnotated による依存性注入(DB・認証・設定)
    3. ルーター共通依存、エンドポイント依存、yield依存(後処理つき)
    4. Pydantic(v2)+pydantic-settings による環境変数ベースの設定管理
    5. 例外ハンドラ/ミドルウェアとの使い分け、バージョニング設計、テストでの依存差し替え
  • 得られる効果
    • 新機能の追加や仕様変更に強い疎結合な設計
    • 共通処理を依存として再利用し、重複のない読みやすいコード
    • テスト時に依存を差し替えられるため、安全で素早い検証が可能

🎯 誰が読んで得をする?(具体像)

  • 学習者Aさん(大学4年・個人開発)
    ルートが増えるほど main.py が膨らんでつらい…。機能ごとに分割し、DB・認証をキレイに渡す方法が欲しい方。
  • 受託開発Bさん(チーム3人)
    認証、レート制限、監査ログなど共通処理を各APIでコピペしていて保守が大変ルーター共通依存にまとめてスッキリさせたい方。
  • SaaS開発Cさん(スタートアップ)
    将来のv2 APIや機能拡張を見据えたバージョニングと設定管理を導入し、テストで依存差し替えを活用したい方。

1. 入口:なぜAPIRouterとDependsなの?

APIRouter は機能単位(例:userstodosauth)でルート群をモジュール化するための仕組み。Depends はエンドポイントが必要とする**共通処理(依存)**を宣言的に受け取る仕組みです。
この2つを組み合わせると、次のようなメリットが得られます。

  • 疎結合:DBセッション、認証済みユーザー、設定などを「引数」として受け取るだけ
  • 再利用性:ルーター全体に共通の依存を一括適用し、コピペを廃止
  • テスト容易性app.dependency_overrides で依存の差し替えができる
  • 見通し:関心ごとごとにフォルダ分割し、保守がラク

要点まとめ

  • APIRouter=モジュール分割、Depends=共通処理の受け渡し
  • 疎結合・再利用・テスト容易性が一気に高まります

2. 最小プロジェクト構成(まずは形から)

以下の構成をベースに進めます。必要に応じてファイルを追加しましょう。

fastapi-arch/
├─ app/
│  ├─ main.py
│  ├─ core/
│  │  ├─ settings.py        # 設定(pydantic-settings)
│  │  ├─ security.py        # 認証関連の依存
│  │  └─ exceptions.py      # 共通例外・ハンドラ
│  ├─ db/
│  │  ├─ session.py         # DBエンジン・SessionLocal・依存
│  │  └─ models.py          # SQLAlchemyモデル
│  ├─ routers/
│  │  ├─ users.py           # APIRouter(ユーザー機能)
│  │  └─ todos.py           # APIRouter(ToDo機能)
│  └─ schemas/
│     ├─ users.py           # Pydanticスキーマ(v2想定)
│     └─ todos.py
└─ tests/
   └─ ...(依存差し替え用のテスト)

用語

  • 依存(Dependency):関数・クラス・設定・DB接続など、エンドポイントが動くために必要なもの。
  • yield依存yield を使い、後処理(クリーンアップ)も同時に書ける依存。

要点まとめ

  • core に横断機能、db にDB関連、routers に機能別ルーター、schemas にI/O定義
  • 役割ごとに分けるだけで、レビューや改修が劇的にラクに

3. 設定管理:pydantic-settingsで環境変数を一元化

Pydantic v2 では設定管理に pydantic-settings を使うのが王道です。

# app/core/settings.py
from pydantic import Field
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My FastAPI"
    env: str = Field("dev", description="環境名: dev/stg/prod")
    database_url: str = "sqlite:///./app.db"
    secret_key: str = Field(..., description="JWT等で使用する秘密鍵")

    class Config:
        env_file = ".env"   # .env を読み込む(任意)
        extra = "ignore"    # 未知の環境変数は無視

# アプリ全体で共有する設定のファクトリ
def get_settings() -> Settings:
    return Settings()
  • ポイントget_settings を依存として注入すれば、テスト時だけ別の設定を簡単に差し替え可能。
  • 値は**必須(…)**や Fielddescription を活用し、自己文書化も叶えます。

要点まとめ

  • 設定は1か所で管理、依存として注入
  • テスト、ステージング、本番で切り替えが容易

4. DBセッションの依存:yieldで後始末まで

アプリ本体から独立させ、yield依存で確実にクローズします。

# app/db/session.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from typing import Generator
from app.core.settings import get_settings

# Engine/SessionLocalを設定依存から作る(実運用はpool設定も)
def get_engine():
    settings = get_settings()
    connect_args = {"check_same_thread": False} if settings.database_url.startswith("sqlite") else {}
    return create_engine(settings.database_url, connect_args=connect_args)

Engine = get_engine()
SessionLocal = sessionmaker(bind=Engine, autoflush=False, autocommit=False)

def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
  • テラコピ防止:DBセッションの生成/破棄はここに集約。
  • テスト時の置換get_dbapp.dependency_overrides で置き換えるだけで、テストDBへ。

要点まとめ

  • DBセッションはyield依存でクローズ保証
  • テストでの差し替えポイントが明確に

5. 認証の依存:認証済みユーザーを「引数」に

認証は依存として引数で受け取るのがFastAPI流。ここでは簡易サンプル(実運用ではJWTなど)。

# app/core/security.py
from fastapi import Depends, HTTPException, status
from typing import Annotated
from app.core.settings import get_settings

# トークンの検証などを行う想定(ここではダミー)
def get_current_user(token: str | None = None, settings = Depends(get_settings)):
    # 例:環境に応じて検証方法を変えることも可能
    if token != "valid":
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized")
    return {"username": "alice", "role": "user"}

# 役割のチェックを重ねる依存
def require_role(role: str):
    def checker(user = Depends(get_current_user)):
        if user.get("role") != role:
            raise HTTPException(status_code=403, detail="Forbidden")
        return user
    return checker

User = Annotated[dict, Depends(get_current_user)]
AdminUser = Annotated[dict, Depends(require_role("admin"))]
  • 多層依存require_roleget_current_user を内部で利用。
  • 型ヒントAnnotated で可読性UP(User/AdminUser など簡易型エイリアス)。

要点まとめ

  • 認証結果を「引数」で受け取ると、疎結合・テスト容易
  • 依存の合成で権限チェックもスッキリ

6. スキーマ:Pydantic v2で from_attributes

# app/schemas/todos.py
from datetime import datetime
from pydantic import BaseModel

class TodoBase(BaseModel):
    title: str
    is_done: bool = False

class TodoCreate(TodoBase):
    pass

class Todo(TodoBase):
    id: int
    created_at: datetime
    class Config:
        from_attributes = True

要点まとめ

  • 入力用(Create/Update)と出力用(Read)を分ける
  • from_attributes=True でORMからの変換をスムーズに

7. ルーター(APIRouter)を定義しよう

7.1 ToDoルーター(ルーター依存+エンドポイント依存)

# app/routers/todos.py
from fastapi import APIRouter, Depends, HTTPException, status
from typing import Annotated, List
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.schemas.todos import Todo, TodoCreate
from app.core.security import User

router = APIRouter(
    prefix="/todos",
    tags=["todos"],
    dependencies=[],  # ここに共通依存を入れると全エンドポイントへ適用
    responses={404: {"description": "Not found"}}
)

DB = Annotated[Session, Depends(get_db)]

@router.post("", response_model=Todo, status_code=201)
def create_todo(payload: TodoCreate, db: DB, user: User):
    # user は認証済みユーザー(依存で取得)
    from app.db.models import Todo as TodoModel
    todo = TodoModel(title=payload.title, is_done=payload.is_done)
    db.add(todo); db.commit(); db.refresh(todo)
    return todo

@router.get("", response_model=List[Todo])
def list_todos(db: DB, user: User):
    from app.db.models import Todo as TodoModel
    return db.query(TodoModel).order_by(TodoModel.id.asc()).all()

7.2 Usersルーター(共通依存をルーターに)

# app/routers/users.py
from fastapi import APIRouter, Depends
from typing import Annotated
from app.core.security import AdminUser

router = APIRouter(
    prefix="/users",
    tags=["users"],
    dependencies=[Depends(lambda: True)]  # 例:監査ログ用ミドル依存など
)

@router.get("/me")
def read_me(user: Annotated[dict, Depends(...)]):
    # 実際は get_current_user を使います。ここはダミー省略。
    return {"username": "alice"}

@router.get("/admin/metrics")
def admin_metrics(_admin: AdminUser):
    return {"status": "ok", "metrics": {"active_users": 123}}

要点まとめ

  • ルーターprefix/tags/responses/dependencies共通属性を集中管理
  • エンドポイントには必要最小限の依存だけを書く

8. main.py:ルーターを集約し、例外ハンドラも登録

# app/main.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from app.routers import todos, users
from app.core.settings import get_settings

app = FastAPI(title="Modular FastAPI")

# 例:共通例外
class DomainError(Exception):
    def __init__(self, message: str):
        self.message = message

@app.exception_handler(DomainError)
def domain_error_handler(_: Request, exc: DomainError):
    return JSONResponse(status_code=400, content={"detail": exc.message})

# ルーターの取り込み(バージョンも付けられる)
app.include_router(todos.router, prefix="/api/v1")
app.include_router(users.router, prefix="/api/v1")

# ルート確認
@app.get("/")
def health(settings = get_settings()):
    return {"app": settings.app_name, "env": settings.env}
  • 例外ハンドラ:業務エラーは例外として上げ、ハンドラで一元変換
  • バージョニング/api/v1include_router 側で付与すると移行がラク
  • 設定依存:トップの /get_settings() を注入する例。

要点まとめ

  • ハンドラで業務エラー→HTTPレスポンスを統一
  • ルーターの取り込み時にバージョンprefixを付けると拡張しやすい

9. ミドルウェア vs 依存:どう使い分ける?

  • ミドルウェア全リクエストに対する前後処理(例:リクエストID、CORS、タイミング計測、ログ集約)
  • 依存ルーター単位/エンドポイント単位での前処理(例:認証、DBセッション、設定、権限チェック)
  • 実務の指針
    • 誰にでも必要」→ミドルウェア
    • この機能だけ必要」→依存
    • 共通だが一部だけ除外したい」→ルーター依存で包む

要点まとめ

  • 作用範囲の広い順:ミドルウェア > ルーター依存 > エンドポイント依存
  • 範囲を絞るほど副作用が小さく、テストしやすい

10. 依存の設計パターン集(コピペで使える)

10.1 ページネーション依存

from fastapi import Query
from typing import Annotated

def pagination(
    limit: int = Query(20, ge=1, le=100),
    offset: int = Query(0, ge=0)
):
    return {"limit": limit, "offset": offset}

Pagination = Annotated[dict, Depends(pagination)]

10.2 トランザクション制御(明示コミット)

from sqlalchemy.orm import Session
from contextlib import contextmanager

@contextmanager
def tx(db: Session):
    try:
        yield
        db.commit()
    except:
        db.rollback()
        raise

10.3 設定のサブセット依存(必要な値だけ)

from typing import TypedDict
class JwtConfig(TypedDict):
    secret_key: str

def get_jwt_config(settings = Depends(get_settings)) -> JwtConfig:
    return {"secret_key": settings.secret_key}

要点まとめ

  • 入力検証(Query/Path/Body)も依存として「束ねる」と綺麗
  • 設定は「必要な断片」にスライスするとテストが軽い

11. ルーター共通依存で重複ゼロへ(実例)

監査ログ・レート制限・権限境界などをルーター丸ごとで付与。

# 例:監査ログのダミー依存
from fastapi import Request
def audit(request: Request):
    # 実際はリクエストID等を使って非同期で集計
    return True

# ルーター定義時に適用
router = APIRouter(
    prefix="/reports",
    tags=["reports"],
    dependencies=[Depends(audit)]
)

要点まとめ

  • 横断的関心ごとはルーター依存で1か所に
  • コピペを根絶して、変更点を1点に集約

12. バージョニング:v1からv2へのスムーズな進化

  • パス方式/api/v1/... → 将来 /api/v2/...include_router で追加
  • 別ルーターrouters_v2/ を作り、新旧APIを並走させて段階的移行
  • スキーマの整合response_model を明示し、互換が崩れた箇所をわかりやすく分離
# main.py(追加)
from app.routers_v2 import todos as todos_v2
app.include_router(todos_v2.router, prefix="/api/v2")

要点まとめ

  • 物理分離(別フォルダ)+論理分離(prefix)で段階的移行
  • 「壊す」変更はv2に隔離し、互換維持

13. 例外とエラーレスポンスの整形(DXを上げる)

  • 共通例外型を定義し、ハンドラで一貫した形に。
  • responses={...} でOpenAPIにも期待形を宣言。
  • 入力バリデーションはPydanticが自動で行うため、業務エラーに集中。
# core/exceptions.py(例)
from fastapi import Request
from fastapi.responses import JSONResponse

class NotEnoughPermission(Exception):
    def __init__(self, action: str): self.action = action

def register_handlers(app):
    @app.exception_handler(NotEnoughPermission)
    def _(req: Request, exc: NotEnoughPermission):
        return JSONResponse(status_code=403, content={"detail": f"Action '{exc.action}' forbidden"})

要点まとめ

  • 例外→レスポンス変換の一元化でクライアント体験UP
  • OpenAPIの responses によりドキュメントと実装を同期

14. テストで依存を差し替える(リグレッションに強く)

APIRouter/Depends の真価はテストで現れます。

# tests/conftest.py(例)
import pytest
from app.main import app
from app.db.session import get_db
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

@pytest.fixture(autouse=True, scope="function")
def override_db():
    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
    TestingSession = sessionmaker(bind=engine, autoflush=False, autocommit=False)
    from app.db.models import Base
    Base.metadata.create_all(engine)

    def _get_db():
        db = TestingSession()
        try: yield db
        finally: db.close()

    app.dependency_overrides[get_db] = _get_db
    yield
    app.dependency_overrides.clear()
  • ポイント:本番DBを触らず、毎回まっさらなDBでテスト。
  • 認証依存も同様に差し替え、ロール別の振る舞いを安全に検証。

要点まとめ

  • 依存を差し替え可能=テスト容易性の高さ
  • DB・認証・設定を自由に入れ替えられる

15. サービス層の薄切り(Routerを“痩せさせる”)

エンドポイントにロジックを書きすぎると、テストや再利用が辛くなります。サービス関数に切り出してRouterを薄く保ちましょう。

# services/todos.py(任意)
from sqlalchemy.orm import Session
from app.db.models import Todo as TodoModel

def create_todo(db: Session, title: str, is_done: bool):
    todo = TodoModel(title=title, is_done=is_done)
    db.add(todo); db.commit(); db.refresh(todo)
    return todo
# routers/todos.py(一部差し替え)
from app.services.todos import create_todo as create_todo_service

@router.post("", response_model=Todo, status_code=201)
def create_todo(payload: TodoCreate, db: DB, user: User):
    return create_todo_service(db, payload.title, payload.is_done)

要点まとめ

  • RouterはI/Oの配線、ロジックはサービス
  • 関数単位のテストが書け、変更に強い

16. サンプル:一気通貫の最小実装(動く雛形)

この記事の要素を凝縮した最小スニペットです。必要なファイルに分割して使ってください。

# app/main.py
from fastapi import FastAPI
from app.routers import todos
from app.core.exceptions import register_handlers

app = FastAPI(title="Scalable API")
app.include_router(todos.router, prefix="/api/v1")
register_handlers(app)

# app/routers/todos.py
from fastapi import APIRouter, Depends, HTTPException, status
from typing import List, Annotated
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.schemas.todos import Todo, TodoCreate
from app.core.security import User

router = APIRouter(prefix="/todos", tags=["todos"])
DB = Annotated[Session, Depends(get_db)]

@router.post("", response_model=Todo, status_code=201)
def create_todo(payload: TodoCreate, db: DB, user: User):
    from app.db.models import Todo as TodoModel
    todo = TodoModel(title=payload.title, is_done=payload.is_done)
    db.add(todo); db.commit(); db.refresh(todo)
    return todo

@router.get("", response_model=List[Todo])
def list_todos(db: DB, user: User):
    from app.db.models import Todo as TodoModel
    return db.query(TodoModel).order_by(TodoModel.id.asc()).all()

要点まとめ

  • 依存は Annotated[T, Depends(...)] で可読性UP
  • ルーターは小さく・薄く、共通処理は依存

17. 品質・運用のコツ(現場で効く小ワザ)

  1. 依存の命名get_dbget_settings のように 動詞+名詞 で統一
  2. 依存の粒度:DB・設定は粗く、パラメータ検証は細かく(合成して使う)
  3. ドキュメント整合response_modelresponses常に明示
  4. 差し替え前提:テストやSandboxでは必ず依存を置換。副作用を限定
  5. APIバージョンinclude_router(..., prefix="/api/v1")初期から適用
  6. 監査・計測:ルーター依存に監査ログ/計測を置いて横断的に収集

要点まとめ

  • 名前・粒度・文書化・差し替え・バージョン・横断処理
  • 6点を守るだけで運用が安定します

18. よくある落とし穴と回避策

症状 原因 回避策
依存が循環参照 互いのモジュールが直接import 依存の抽象(プロトコル/TypedDict)インターフェイス層を切る
DBセッションが閉じられない yield依存なし/例外時に後処理漏れ yield 依存で必ずclose、サービス層での例外は上に投げる
ルーターが太る その場でロジック全部書く サービス関数へ切り出し、RouterはI/Oだけ
設定が散らばる 各所で環境変数を読む pydantic-settings で一元化、依存注入
v2移行が混乱 v1と混在し境界不明 フォルダ分離/api/v2 prefixで明確化

要点まとめ

  • 依存は一方向に、後処理はyieldで、Routerは薄く
  • 設定は1か所、バージョンは分離

19. まとめ(今日から“拡張に強い”骨格へ♡)

  • APIRouter でモジュール分割し、共通属性とprefix/tags/responsesを集中管理。
  • DependsAnnotated で、DB・認証・設定・検証を疎結合に注入。
  • yield依存で後処理を保証し、テストでは依存差し替えで安全に検証。
  • pydantic-settings による一元設定、例外ハンドラによるDX向上、バージョニングで進化に強く。

この骨格があれば、新しい機能はルーターを1枚増やすだけ。共通処理は依存を1つ追加するだけ。
コードは自然と整理され、レビューもテストも軽くなります。今日の小さな整備が、明日の大きな安心につながります。わたしも全力で応援していますね♡


付録A:対象読者とインパクトの詳細

  • 個人開発者:APIRouter+Dependsで読みやすさ差し替えやすさが向上。小規模でも撤退コストが減り、学習効果が高い。
  • 小規模チーム:横断処理(監査・認証)をルーター依存に集約し、変更の影響範囲が明確に。レビューとオンボーディングが容易。
  • 成長中SaaS:設定の一元化・バージョニング・例外ハンドラで安全に拡張。テストの依存差し替えでCIの安定に直結。

投稿者 greeden

コメントを残す

メールアドレスが公開されることはありません。 が付いている欄は必須項目です

日本語が含まれない投稿は無視されますのでご注意ください。(スパム対策)