FastAPI 大規模アプリ設計：APIRouter・依存性注入の階層化・プロジェクト構成で『保守できるAPI』を作る

# app/routers/users.py
from fastapi import APIRouter

router = APIRouter()    # この機能（users）のルートを束ねる入れ物

@router.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]

@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):
    return {"username": username}

.
├── app                  # "app" は Python パッケージ
│   ├── __init__.py      # これで "app" がパッケージになる
│   ├── main.py          # アプリ本体（FastAPI() と include_router）
│   ├── dependencies.py  # 共有の依存（認証・共通ヘッダ検証など）
│   ├── routers          # "routers" はサブパッケージ
│   │   ├── __init__.py
│   │   ├── items.py     # items 機能のルート
│   │   └── users.py     # users 機能のルート
│   └── internal         # 社内/管理用など外に出さない機能
│       ├── __init__.py
│       └── admin.py

# app/dependencies.py
from typing import Annotated
from fastapi import Header, HTTPException

async def get_token_header(x_token: Annotated[str, Header()]):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")

async def get_query_token(token: str):
    if token != "jessica":
        raise HTTPException(status_code=400, detail="No Jessica token provided")

from ..dependencies import get_token_header
#   ^^  二つのドット = 親パッケージ（app）まで上がってから dependencies を見る
#   一つのドット（.）= 同じパッケージ内

# app/main.py
from fastapi import Depends, FastAPI
from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users

app = FastAPI(dependencies=[Depends(get_query_token)])   # ← 全体に効く依存（第4章）

app.include_router(users.router)
app.include_router(items.router)

# admin は「組み込み時に」prefix・タグ・追加の依存・追加レスポンスを上書き付与する
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],     # /admin 配下すべてにトークン検証を強制
    responses={418: {"description": "I'm a teapot"}},
)

@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}

parent.include_router(child_router)   # APIRouter を APIRouter に入れ子にできる

# 全リクエストに先立ってヘッダ検証を走らせる（一箇所で全体に効く）
app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])

@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    # verify_token / verify_key は必ず実行される（失敗すれば例外で弾かれる）。
    # ただし戻り値はハンドラに渡らない（使わないので引数にも現れない）。
    return [{"item": "Foo"}, {"item": "Bar"}]

from fastapi import APIRouter, Security
from app.core.security import get_current_active_user   # 認証ガイド参照

# /admin 配下の全ルートに admin スコープを強制（DRY・抜け漏れ防止）
admin_router = APIRouter(
    prefix="/admin",
    tags=["admin"],
    dependencies=[Security(get_current_active_user, scopes=["admin"])],
)

@admin_router.get("/reports")     # ← ここに認可を書かなくても admin が要求される
async def list_reports():
    ...

# app/repositories/user_repository.py —— I/O のみ。SQLの詳細はここに閉じる
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import User

class UserRepository:
    def __init__(self, db: AsyncSession) -> None:
        self._db = db

    async def get(self, user_id: int) -> User | None:
        return await self._db.get(User, user_id)

    async def add(self, user: User) -> User:
        self._db.add(user)
        await self._db.flush()
        return user

# app/services/user_service.py —— 業務ルールのみ。HTTPもSQLも登場しない（純粋でテストしやすい）
from app.repositories.user_repository import UserRepository
from app.schemas.user import UserCreate, UserPublic

class UserService:
    def __init__(self, repo: UserRepository) -> None:
        self._repo = repo

    async def register(self, payload: UserCreate) -> UserPublic:
        # 例：業務ルール（重複チェック・初期ロール付与など）はここに集約する
        user = await self._repo.add(payload.to_model())
        return UserPublic.model_validate(user)

# app/api/deps.py —— 「配線（DI）」を一箇所に集約する
from typing import Annotated
from fastapi import Depends
from app.core.db import DbDep                       # lifespan で作った session を注入（本番運用ガイド参照）
from app.repositories.user_repository import UserRepository
from app.services.user_service import UserService

def get_user_service(db: DbDep) -> UserService:
    # repository を生成して service に注入する。これがDIの結節点。
    return UserService(UserRepository(db))

UserServiceDep = Annotated[UserService, Depends(get_user_service)]

# app/api/v1/routers/users.py —— router は薄い。HTTP境界の翻訳だけ
from fastapi import APIRouter, status
from app.api.deps import UserServiceDep
from app.schemas.user import UserCreate, UserPublic

router = APIRouter()

@router.post("/", response_model=UserPublic, status_code=status.HTTP_201_CREATED)
async def create_user(payload: UserCreate, service: UserServiceDep) -> UserPublic:
    # router の仕事は「受け取って、serviceに渡して、返す」だけ。業務ルールはここに無い。
    return await service.register(payload)

app/
├── main.py            # create_app・include_router・lifespan・middleware（第8章）
├── core/              # 全層から参照される土台（循環の起点にしない）
│   ├── config.py      # Settings（pydantic-settings）
│   ├── db.py          # engine・session factory・get_db（DbDep）
│   └── security.py    # 認証の依存（get_current_user 等）
├── api/
│   ├── deps.py        # DIの結節点（service生成・型エイリアス）
│   └── v1/
│       ├── router.py  # v1の全routerを束ねる
│       └── routers/   # users.py・items.py …（HTTP境界）
├── services/          # ビジネスロジック
├── repositories/      # I/O
├── models/            # ORMモデル
└── schemas/           # Pydanticスキーマ（入力/出力）

main → api(routers) → services → repositories → models / schemas
                  ↘ core(config / db / security) ← 誰からでも参照してよい土台

from __future__ import annotations
from typing import TYPE_CHECKING

if TYPE_CHECKING:           # 型チェック時だけ評価され、実行時には import されない
    from app.services.user_service import UserService

def handler(service: "UserService") -> None: ...

# app/api/v1/router.py —— v1 の全機能を1つの router に束ねる
from fastapi import APIRouter
from .routers import users, items

api_router = APIRouter()
api_router.include_router(users.router, prefix="/users", tags=["users"])
api_router.include_router(items.router, prefix="/items", tags=["items"])

# app/main.py —— /api/v1 配下にまとめて組み込む
from app.api.v1.router import api_router as v1_router
# from app.api.v2.router import api_router as v2_router   # v2 を足しても v1 は無傷

app.include_router(v1_router, prefix="/api/v1")
# app.include_router(v2_router, prefix="/api/v2")

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.core.config import Settings, get_settings
from app.core.lifespan import lifespan          # DBプール等の確保（本番運用ガイド参照）
from app.core.errors import register_exception_handlers
from app.api.v1.router import api_router as v1_router

tags_metadata = [
    {"name": "users", "description": "ユーザーの作成・取得。**ログイン**もここ。"},
    {"name": "items", "description": "アイテム管理。"},
]

def create_app(settings: Settings | None = None) -> FastAPI:
    settings = settings or get_settings()
    app = FastAPI(
        title=settings.app_name,
        version=settings.version,
        lifespan=lifespan,
        openapi_tags=tags_metadata,
        # 本番では docs/openapi を閉じて API 構造の露出を絞る（後述）
        docs_url="/docs" if settings.docs_enabled else None,
        redoc_url=None,
        openapi_url="/openapi.json" if settings.docs_enabled else None,
    )
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,     # "*" は使わない（認証ガイド参照）
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    register_exception_handlers(app)             # 例外→HTTPの翻訳を一箇所に
    app.include_router(v1_router, prefix="/api/v1")
    return app

app = create_app()       # ASGIサーバ（fastapi run / uvicorn）が参照するエントリポイント

app = FastAPI(
    title="ChimichangApp",
    summary="一行サマリ（OpenAPI 3.1 / FastAPI 0.99.0+）",
    description="Markdownが使える詳細説明。**太字**やリストも可。",
    version="1.4.0",
    contact={"name": "API Support", "url": "https://example.com/contact", "email": "api@example.com"},
    license_info={"name": "Apache 2.0", "url": "https://www.apache.org/licenses/LICENSE-2.0.html"},
    openapi_tags=tags_metadata,    # ← この **リストの順序が docs の表示順**になる
)

fastapi dev app/main.py     # 開発：自動リロード（既定は 127.0.0.1:8000）
fastapi run app/main.py     # 本番：リロード無し（既定は 0.0.0.0:8000）

[tool.fastapi]
entrypoint = "app.main:app"   # from app.main import app と同義

# tests/test_users.py
import pytest
from fastapi.testclient import TestClient
from app.main import create_app
from app.api.deps import get_user_service

@pytest.fixture
def client():
    app = create_app(test_settings())             # 本番から隔離した新鮮なアプリ
    # service を「フェイク」に差し替える → DB を立てずに router の契約だけ検証できる
    app.dependency_overrides[get_user_service] = lambda: FakeUserService()
    yield TestClient(app)
    app.dependency_overrides.clear()              # テスト間の汚染を防ぐ

def test_create_user_returns_201(client):
    res = client.post("/api/v1/users/", json={"email": "a@example.com", "display_name": "A"})
    assert res.status_code == 201
    assert "id" in res.json()