fastapi-clean-architecture

FastAPI クリーンアーキテクチャスキル

概要

このスキルでは、エンタープライズ向け金融システムの本番環境でのパターンに基づき、クリーンアーキテクチャの原則に従って FastAPI アプリケーションを構築する方法を説明します。適切なレイヤー分離、依存性注入、リポジトリパターン、包括的なテスト戦略を重視します。

このスキルを使用するタイミング

• クリーンアーキテクチャが必要な新しい FastAPI プロジェクトを始めるとき
• 保守性を高めるために既存の FastAPI コードをリファクタリングするとき
• FastAPI でドメイン駆動設計を実装するとき
• テスト可能でスケーラブルなバックエンドサービスを構築するとき
• 複数のデータソース（PostgreSQL、Redis、外部 API）を統合するとき

アーキテクチャの基本原則

3 層アーキテクチャ

src/
├── api/              # API Layer (Controllers, Routes, DTOs)
├── domain/           # Domain Layer (Business Logic, Entities, Services)
└── infra/            # Infrastructure Layer (Database, External Services)

各層の責務：

1. API 層（src/api/）

   • HTTP エンドポイントとルーティング
   • リクエスト／レスポンス DTO（Pydantic モデル）
   • 入力の検証
   • 認証／認可ミドルウェア
   • エラーハンドリングと HTTP ステータスコード

2. ドメイン層（src/domain/）

   • ビジネスエンティティと値オブジェクト
   • サービスインターフェース（抽象クラス）
   • ビジネスルールと検証
   • ドメイン例外
   • 純粋なビジネスロジック（フレームワークに依存しない）

3. インフラストラクチャ層（src/infra/）

   • データベースリポジトリ（SQLAlchemy）
   • 外部 API アダプター
   • キャッシュの実装（Redis）
   • ファイルストレージアダプター
   • 具体的なサービスの実装

依存関係の流れ

API Layer → Domain Layer ← Infrastructure Layer

重要なルール：

• API はドメインに依存します（ドメインのサービス／エンティティをインポートします）
• インフラストラクチャはドメインのインターフェースを実装します
• ドメインから API やインフラストラクチャをインポートしてはいけません
• 依存性注入を使って、すべてを連携させます

プロジェクト構成のテンプレート

project-name/
├── src/
│   ├── api/
│   │   ├── path/              # Route modules
│   │   │   ├── users.py
│   │   │   ├── auth.py
│   │   │   └── financial.py
│   │   ├── middlewares/       # Custom middleware
│   │   │   ├── auth.py
│   │   │   └── error_handler.py
│   │   └── schemas/           # Request/Response DTOs
│   │       ├── user.py
│   │       └── financial.py
│   ├── domain/
│   │   ├── modules/           # Business modules
│   │   │   ├── user/
│   │   │   │   ├── entity.py       # User entity
│   │   │   │   ├── service.py      # IUserService (abstract)
│   │   │   │   ├── repository.py   # IUserRepository (abstract)
│   │   │   │   └── exceptions.py   # Domain exceptions
│   │   │   └── financial/
│   │   │       ├── entity.py
│   │   │       ├── service.py
│   │   │       └── value_objects.py
│   │   └── shared/            # Shared domain logic
│   │       ├── base_entity.py
│   │       └── exceptions.py
│   ├── infra/
│   │   ├── database/
│   │   │   ├── alchemist/     # SQLAlchemy implementations
│   │   │   │   ├── modules/
│   │   │   │   │   ├── user/
│   │   │   │   │   │   └── repository.py  # UserRepository
│   │   │   │   │   └── financial/
│   │   │   │   │       └── repository.py
│   │   │   │   └── session.py
│   │   │   └── migrations/    # Database migrations
│   │   ├── adapters/          # External service adapters
│   │   │   ├── auth/
│   │   │   │   └── sso_adapter.py
│   │   │   └── payment/
│   │   │       └── payment_gateway.py
│   │   ├── cache/             # Redis implementations
│   │   │   └── redis_cache.py
│   │   └── services/          # Service implementations
│   │       ├── user_service.py
│   │       └── financial_service.py
│   ├── config/
│   │   ├── settings.py        # Environment configuration
│   │   ├── dependency.py      # DI Container
│   │   └── database.py        # DB configuration
│   └── main.py                # FastAPI app entry point
├── tests/
│   ├── api/                   # API integration tests
│   ├── domain/                # Domain unit tests
│   ├── infra/                 # Infrastructure tests
│   └── conftest.py            # Pytest fixtures
├── pyproject.toml
├── Dockerfile
├── docker-compose.yml
└── README.md

実装パターン

1. ドメインエンティティパターン

# src/domain/modules/user/entity.py
from dataclasses import dataclass
from datetime import datetime

@dataclass
class User:
    """Domain entity representing a user.

    Pure business logic with no framework dependencies.
    """
    id: int | None
    cpf: str
    name: str
    email: str
    created_at: datetime
    is_active: bool = True

    def deactivate(self) -> None:
        """Business rule: Deactivate user account."""
        if not self.is_active:
            raise UserAlreadyInactiveError(f"User {self.cpf} is already inactive")
        self.is_active = False

    def validate_cpf(self) -> bool:
        """Business rule: Validate CPF format."""
        cleaned = ''.join(filter(str.isdigit, self.cpf))
        return len(cleaned) == 11 and not all(c == cleaned[0] for c in cleaned)

2. リポジトリインターフェースパターン

# src/domain/modules/user/repository.py
from abc import ABC, abstractmethod
from typing import List

from src.domain.modules.user.entity import User

class IUserRepository(ABC):
    """Abstract repository interface in domain layer.

    Defines contract without implementation details.
    """

    @abstractmethod
    async def get_by_id(self, user_id: int) -> User | None:
        """Retrieve user by ID."""
        pass

    @abstractmethod
    async def get_by_cpf(self, cpf: str) -> User | None:
        """Retrieve user by CPF."""
        pass

    @abstractmethod
    async def create(self, user: User) -> User:
        """Create new user."""
        pass

    @abstractmethod
    async def update(self, user: User) -> User:
        """Update existing user."""
        pass

    @abstractmethod
    async def list_active(self, limit: int = 100) -> List[User]:
        """List all active users."""
        pass

3. サービスインターフェースパターン

# src/domain/modules/user/service.py
from abc import ABC, abstractmethod
from typing import List

from src.domain.modules.user.entity import User

class IUserService(ABC):
    """Abstract service interface defining business operations."""

    @abstractmethod
    async def register_user(self, cpf: str, name: str, email: str) -> User:
        """Register new user with validation."""
        pass

    @abstractmethod
    async def deactivate_user(self, cpf: str) -> User:
        """Deactivate user account."""
        pass

    @abstractmethod
    async def get_user_profile(self, cpf: str) -> User:
        """Get user profile by CPF."""
        pass

4. 具体的なリポジトリの実装

# src/infra/database/alchemist/modules/user/repository.py
from typing import List
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession

from src.domain.modules.user.entity import User
from src.domain.modules.user.repository import IUserRepository
from src.infra.database.alchemist.models import UserModel

class UserRepository(IUserRepository):
    """SQLAlchemy implementation of user repository.

    Handles database operations and entity mapping.
    """

    def __init__(self, session: AsyncSession):
        self._session = session

    async def get_by_id(self, user_id: int) -> User | None:
        """Retrieve user by ID."""
        result = await self._session.execute(
            select(UserModel).where(UserModel.id == user_id)
        )
        model = result.scalar_one_or_none()
        return self._to_entity(model) if model else None

    async def get_by_cpf(self, cpf: str) -> User | None:
        """Retrieve user by CPF."""
        result = await self._session.execute(
            select(UserModel).where(UserModel.cpf == cpf)
        )
        model = result.scalar_one_or_none()
        return self._to_entity(model) if model else None

    async def create(self, user: User) -> User:
        """Create new user."""
        model = UserModel(
            cpf=user.cpf,
            name=user.name,
            email=user.email,
            is_active=user.is_active,
        )
        self._session.add(model)
        await self._session.flush()
        await self._session.refresh(model)
        return self._to_entity(model)

    async def update(self, user: User) -> User:
        """Update existing user."""
        result = await self._session.execute(
            select(UserModel).where(UserModel.id == user.id)
        )
        model = result.scalar_one()
        model.name = user.name
        model.email = user.email
        model.is_active = user.is_active
        await self._session.flush()
        await self._session.refresh(model)
        return self._to_entity(model)

    async def list_active(self, limit: int = 100) -> List[User]:
        """List all active users."""
        result = await self._session.execute(
            select(UserModel)
            .where(UserModel.is_active == True)  # noqa: E712
            .limit(limit)
        )
        models = result.scalars().all()
        return [self._to_entity(model) for model in models]

    def _to_entity(self, model: UserModel) -> User:
        """Convert database model to domain entity."""
        return User(
            id=model.id,
            cpf=model.cpf,
            name=model.name,
            email=model.email,
            created_at=model.created_at,
            is_active=model.is_active,
        )

5. 具体的なサービスの実装

# src/infra/services/user_service.py
from src.domain.modules.user.entity import User
from src.domain.modules.user.service import IUserService
from src.domain.modules.user.repository import IUserRepository
from src.domain.modules.user.exceptions import (
    UserAlreadyExistsError,
    UserNotFoundError,
    InvalidCPFError,
)

class UserService(IUserService):
    """Concrete implementation of user service.

    Orchestrates business logic using repository.
    """

    def __init__(self, user_repository: IUserRepository):
        self._repository = user_repository

    async def register_user(self, cpf: str, name: str, email: str) -> User:
        """Register new user with validation."""
        # Check if user already exists
        existing_user = await self._repository.get_by_cpf(cpf)
        if existing_user:
            raise UserAlreadyExistsError(f"User with CPF {cpf} already exists")

        # Create and validate entity
        user = User(
            id=None,
            cpf=cpf,
            name=name,
            email=email,
            created_at=datetime.now(),
            is_active=True,
        )

        if not user.validate_cpf():
            raise InvalidCPFError(f"Invalid CPF: {cpf}")

        # Persist
        return await self._repository.create(user)

    async def deactivate_user(self, cpf: str) -> User:
        """Deactivate user account."""
        user = await self._repository.get_by_cpf(cpf)
        if not user:
            raise UserNotFoundError(f"User with CPF {cpf} not found")

        user.deactivate()  # Business logic in entity
        return await self._repository.update(user)

    async def get_user_profile(self, cpf: str) -> User:
        """Get user profile by CPF."""
        user = await self._repository.get_by_cpf(cpf)
        if not user:
            raise UserNotFoundError(f"User with CPF {cpf} not found")
        return user

6. 依存性注入コンテナー

# src/config/dependency.py
from dependency_injector import containers, providers
from dependency_injector.wiring import Provide, inject

from src.infra.database.alchemist.session import get_session
from src.infra.database.alchemist.modules.user.repository import UserRepository
from src.infra.services.user_service import UserService

class Container(containers.DeclarativeContainer):
    """Dependency injection container.

    Wires together all dependencies.
    """

    wiring_config = containers.WiringConfiguration(
        modules=[
            "src.api.path.users",
            "src.api.path.auth",
        ]
    )

    # Database
    db_session = providers.Resource(get_session)

    # Repositories
    user_repository = providers.Factory(
        UserRepository,
        session=db_session,
    )

    # Services
    user_service = providers.Factory(
        UserService,
        user_repository=user_repository,
    )

7. API エンドポイントの実装

# src/api/path/users.py
from fastapi import APIRouter, Depends, status
from dependency_injector.wiring import inject, Provide

from src.api.schemas.user import UserCreateRequest, UserResponse
from src.domain.modules.user.service import IUserService
from src.domain.modules.user.exceptions import UserAlreadyExistsError, InvalidCPFError
from src.config.dependency import Container

router = APIRouter(prefix="/v1/users", tags=["users"])

@router.post(
    "/",
    response_model=UserResponse,
    status_code=status.HTTP_201_CREATED,
)
@inject
async def create_user(
    request: UserCreateRequest,
    user_service: IUserService = Depends(Provide[Container.user_service]),
):
    """Create new user endpoint.

    Delegates to service layer for business logic.
    """
    try:
        user = await user_service.register_user(
            cpf=request.cpf,
            name=request.name,
            email=request.email,
        )
        return UserResponse.from_entity(user)
    except UserAlreadyExistsError as e:
        raise HTTPException(status_code=409, detail=str(e))
    except InvalidCPFError as e:
        raise HTTPException(status_code=400, detail=str(e))

@router.get("/{cpf}", response_model=UserResponse)
@inject
async def get_user(
    cpf: str,
    user_service: IUserService = Depends(Provide[Container.user_service]),
):
    """Get user by CPF."""
    user = await user_service.get_user_profile(cpf)
    return UserResponse.from_entity(user)

8. Pydantic DTO（リクエスト／レスポンス）

# src/api/schemas/user.py
from datetime import datetime
from pydantic import BaseModel, EmailStr, Field

from src.domain.modules.user.entity import User

class UserCreateRequest(BaseModel):
    """Request DTO for user creation."""
    cpf: str = Field(..., min_length=11, max_length=14)
    name: str = Field(..., min_length=3, max_length=100)
    email: EmailStr

class UserResponse(BaseModel):
    """Response DTO for user data."""
    id: int
    cpf: str
    name: str
    email: str
    created_at: datetime
    is_active: bool

    @classmethod
    def from_entity(cls, user: User) -> "UserResponse":
        """Convert domain entity to DTO."""
        return cls(
            id=user.id,
            cpf=user.cpf,
            name=user.name,
            email=user.email,
            created_at=user.created_at,
            is_active=user.is_active,
        )

テスト戦略

ユニットテスト（ドメイン層）

# tests/domain/modules/user/test_entity.py
import pytest
from src.domain.modules.user.entity import User
from src.domain.modules.user.exceptions import UserAlreadyInactiveError

def test_user_deactivation():
    """Test user deactivation business rule."""
    user = User(
        id=1,
        cpf="12345678901",
        name="Test User",
        email="[email protected]",
        created_at=datetime.now(),
        is_active=True,
    )

    user.deactivate()
    assert user.is_active is False

def test_user_already_inactive_raises_error():
    """Test deactivating already inactive user raises error."""
    user = User(
        id=1,
        cpf="12345678901",
        name="Test User",
        email="[email protected]",
        created_at=datetime.now(),
        is_active=False,
    )

    with pytest.raises(UserAlreadyInactiveError):
        user.deactivate()

def test_cpf_validation():
    """Test CPF validation logic."""
    user = User(
        id=1,
        cpf="12345678901",
        name="Test User",
        email="[email protected]",
        created_at=datetime.now(),
    )

    assert user.validate_cpf() is True

    user.cpf = "11111111111"  # All same digits
    assert user.validate_cpf() is False

サービステスト（インフラストラクチャ層）

# tests/infra/services/test_user_service.py
import pytest
from unittest.mock import AsyncMock

from src.infra.services.user_service import UserService
from src.domain.modules.user.entity import User
from src.domain.modules.user.exceptions import UserAlreadyExistsError

@pytest.fixture
def mock_repository():
    """Mock user repository."""
    return AsyncMock()

@pytest.fixture
def user_service(mock_repository):
    """User service with mocked repository."""
    return UserService(user_repository=mock_repository)

@pytest.mark.asyncio
async def test_register_user_success(user_service, mock_repository):
    """Test successful user registration."""
    mock_repository.get_by_cpf.return_value = None
    mock_repository.create.return_value = User(
        id=1,
        cpf="12345678901",
        name="Test User",
        email="[email protected]",
        created_at=datetime.now(),
        is_active=True,
    )

    user = await user_service.register_user(
        cpf="12345678901",
        name="Test User",
        email="[email protected]",
    )

    assert user.id == 1
    assert user.cpf == "12345678901"
    mock_repository.create.assert_called_once()

@pytest.mark.asyncio
async def test_register_user_already_exists(user_service, mock_repository):
    """Test registering existing user raises error."""
    mock_repository.get_by_cpf.return_value = User(
        id=1,
        cpf="12345678901",
        name="Existing User",
        email="[email protected]",
        created_at=datetime.now(),
        is_active=True,
    )

    with pytest.raises(UserAlreadyExistsError):
        await user_service.register_user(
            cpf="12345678901",
            name="Test User",
            email="[email protected]",
        )

統合テスト（API 層）

# tests/api/test_users.py
import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_create_user_endpoint(client: AsyncClient):
    """Test user creation endpoint."""
    response = await client.post(
        "/v1/users/",
        json={
            "cpf": "12345678901",
            "name": "Test User",
            "email": "[email protected]",
        },
    )

    assert response.status_code == 201
    data = response.json()
    assert data["cpf"] == "12345678901"
    assert data["name"] == "Test User"

@pytest.mark.asyncio
async def test_create_duplicate_user_returns_409(client: AsyncClient):
    """Test creating duplicate user returns conflict."""
    user_data = {
        "cpf": "12345678901",
        "name": "Test User",
        "email": "[email protected]",
    }

    # Create first user
    await client.post("/v1/users/", json=user_data)

    # Try to create duplicate
    response = await client.post("/v1/users/", json=user_data)
    assert response.status_code == 409

データベースセッション管理

# src/infra/database/alchemist/session.py
from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    create_async_engine,
    async_sessionmaker,
)

from src.config.settings import app_settings

# Engine configuration
engine = create_async_engine(
    app_settings.DATABASE_URL,
    echo=app_settings.DEBUG,
    pool_size=10,
    max_overflow=20,
    pool_recycle=3600,
)

# Session factory
AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False,
)

@asynccontextmanager
async def get_session() -> AsyncSession:
    """Get database session with automatic cleanup.

    Usage with dependency injection:
        session = Depends(get_session)
    """
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

設定管理

# src/config/settings.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class AppSettings(BaseSettings):
    """Application configuration from environment variables."""

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=True,
    )

    # Application
    APP_NAME: str = "FastAPI Clean Architecture"
    DEBUG: bool = False
    API_VERSION: str = "v1"

    # Database
    DATABASE_URL: str
    DB_POOL_SIZE: int = 10
    DB_MAX_OVERFLOW: int = 20

    # Redis
    REDIS_URL: str = "redis://localhost:6379/0"
    REDIS_TTL: int = 3600

    # Security
    JWT_SECRET_KEY: str
    JWT_ALGORITHM: str = "RS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60

    # CORS
    CORS_ORIGINS: list[str] = ["*"]
    CORS_CREDENTIALS: bool = True

app_settings = AppSettings()

メインアプリケーションのセットアップ

# src/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from src.config.settings import app_settings
from src.config.dependency import Container
from src.api.path import users, auth

def create_app() -> FastAPI:
    """Create and configure FastAPI application."""

    # Initialize DI container
    container = Container()

    # Create app
    app = FastAPI(
        title=app_settings.APP_NAME,
        version=app_settings.API_VERSION,
        debug=app_settings.DEBUG,
    )

    # Wire container
    app.container = container

    # CORS middleware
    app.add_middleware(
        CORSMiddleware,
        allow_origins=app_settings.CORS_ORIGINS,
        allow_credentials=app_settings.CORS_CREDENTIALS,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    # Include routers
    app.include_router(users.router)
    app.include_router(auth.router)

    return app

app = create_app()

@app.get("/health")
async def health_check():
    """Health check endpoint."""
    return {"status": "healthy"}

ベストプラクティスのチェックリスト

アーキテクチャ

• ✅ ドメイン層は API やインフラストラクチャに一切依存しません
• ✅ すべてのデータベースロジックをリポジトリ内にカプセル化します
• ✅ ビジネスロジックはドメインエンティティとサービスに記述します
• ✅ インフラストラクチャはドメインインターフェースを実装します
• ✅ API 層では HTTP に関する処理のみを扱います

依存性注入

• ✅ IoC コンテナーには dependency-injector を使用します
• ✅ すべての依存関係をコンテナー経由で連携させます
• ✅ 具体的な実装ではなく、インターフェースを注入します
• ✅ エンドポイントに @inject デコレーターを使用します
• ✅ すべての API モジュールで配線を設定します

データベース

• ✅ SQLAlchemy 2.0 の非同期パターンを使用します
• ✅ ORM モデルとドメインエンティティを分離します
• ✅ マッパーメソッド（_to_entity）を実装します
• ✅ コネクションプーリングを使用します
• ✅ トランザクションを適切に処理します（コミット／ロールバック）

テスト

• ✅ ドメイン層（純粋なロジック）のカバレッジを 100% にします
• ✅ サービステストではリポジトリをモックします
• ✅ 非同期メソッドには AsyncMock を使用します
• ✅ エンドポイントの統合テストを実施します
• ✅ 統合テスト用のテストデータベースを分けて用意します

命名規則

• ✅ エンティティ：PascalCase（User、Order）
• ✅ サービス：I{Name}Service（インターフェース）、{Name}Service（実装）
• ✅ リポジトリ：I{Name}Repository（インターフェース）、{Name}Repository（実装）
• ✅ DTO：{Name}Request、{Name}Response
• ✅ 該当する場合は、データベースの列名に ptBR の名前を使用します

エラーハンドリング

• ✅ ビジネスルール違反にはドメイン例外を使用します
• ✅ HTTP 例外は API 層でのみ使用します
• ✅ 適切なステータスコード（400、404、409、500）を使用します
• ✅ 意味のあるエラーメッセージを設定します
• ✅ グローバル例外ハンドラーを設定します

よくある落とし穴と回避方法

1. ドメイン層でのインフラストラクチャのインポート

   • ❌ ドメイン層では SQLAlchemy モデルを絶対にインポートしないでください
   • ✅ マッパー関数を使ってレイヤー間を変換してください

2. API 層におけるビジネスロジック

   • ❌ エンドポイントにバリデーションやビジネスルールを絶対に記述しないでください
   • ✅ すべてのロジックをサービスまたはエンティティに移してください

3. 強い結合

   • ❌ 依存関係を直接インスタンス化しないでください
   • ✅ すべての箇所で依存性注入を使用してください

4. 貧血エンティティ

   • ❌ エンティティを単なるデータコンテナーとして使わないでください
   • ✅ エンティティに振る舞いとバリデーションを実装してください

5. リポジトリの境界漏れ

   • ❌ リポジトリの外部に SQLAlchemy クエリを公開しないでください
   • ✅ ドメインエンティティのみを返してください

6. 不適切なトランザクション管理

   • ❌ リポジトリ内でコミットやロールバックを行わないでください
   • ✅ サービスまたはエンドポイントのレベルでトランザクションを管理してください

移行ガイド（レガシー → クリーンアーキテクチャ）

ステップ 1：ドメイン層を作成する

1. データベースモデルからビジネスエンティティを抽出します
2. リポジトリインターフェースを定義します
3. サービスインターフェースを定義します
4. ビジネスロジックをエンティティ／サービスに移します

ステップ 2：インフラストラクチャ層を作成する

1. SQLAlchemy を使ってリポジトリを実装します
2. サービスの実装を作成します
3. データベースモデルをエンティティから分離します

ステップ 3：API 層をリファクタリングする

1. リクエスト／レスポンス DTO を作成します
2. サービスを使用するようエンドポイントを更新します
3. データベースへの直接アクセスを削除します
4. 依存性注入を追加します

ステップ 4：テスト

1. ドメイン層の単体テストを作成します
2. モックリポジトリを使ったサービステストを追加します
3. エンドポイントの統合テストを作成します
4. 100％のカバレッジを確保します

参考資料

• Robert C. Martin 著『Clean Architecture』
• FastAPI ドキュメント
• Dependency Injector
• SQLAlchemy 2.0
• Pydantic

本番環境での例

このスキルは、以下のパターンに基づいています。

• GEFIN Backend：595 件以上のテストを備えた財務管理システム
• クリーンアーキテクチャ：ドメイン駆動設計の原則
• エンタープライズのベストプラクティス：スケーラビリティ、保守性、テスト容易性