アーキテクチャパターン

Clean Architecture、Hexagonal Architecture、Domain-Driven Design などの実証済みバックエンドアーキテクチャパターンをマスターして、保守性が高く、テスト可能でスケーラブルなシステムを構築しましょう。

入力: アーキテクチャを設計するサービス境界またはモジュール 出力: 明確な依存規則、インターフェース定義、テスト境界を持つレイヤード構造

このスキルを使う場面

• ゼロからバックエンドサービスまたはマイクロサービスを設計する
• ビジネスロジックが ORMモデルや HTTP関連の関心事と絡み合ったモノリシックアプリケーションをリファクタリングする
• システムをサービスに分割する前に Bounded Context を確立する
• インフラストラクチャコードがドメインレイヤーに流出している依存関係サイクルをデバッグする
• ユースケーステストが実行中のデータベースを必要としないテスト可能なコードベースを作成する
• Domain-Driven Design の戦術的パターン（集約、値オブジェクト、ドメインイベント）を実装する

コア概念

1. Clean Architecture（Uncle Bob）

レイヤー（依存関係は内側に流れる）:

• Entities: コアビジネスモデル、フレームワークのインポートなし
• Use Cases: アプリケーションビジネスルール、エンティティを調整
• Interface Adapters: コントローラー、プレゼンター、ゲートウェイ — ユースケースと外部フォーマット間の変換
• Frameworks & Drivers: UI、データベース、外部サービス — すべて最外輪に配置

主要な原則:

• 依存関係は内側にのみ向く。内側のレイヤーは外側のレイヤーについて何も知らない
• ビジネスロジックはフレームワーク、データベース、配信メカニズムから独立している
• すべてのレイヤー境界は抽象インターフェースを経由して交差する
• UI、データベース、外部サービスなしでテスト可能

2. Hexagonal Architecture（Ports and Adapters）

コンポーネント:

• Domain Core: ビジネスロジックがここに存在、フレームワークフリー
• Ports: コアが外部世界と相互作用する方法を定義する抽象インターフェース（駆動型と被駆動型）
• Adapters: ポートの具体的な実装（PostgreSQL アダプター、Stripe アダプター、REST アダプター）

メリット:

• コアに触れることなく実装を入れ替える（例：PostgreSQL から DynamoDB に変更）
• テストでインメモリアダプターを使用 — Docker は不要
• 技術的な決定を端に遅延させる

3. Domain-Driven Design（DDD）

戦略的パターン:

• Bounded Contexts: 1つのサブドメイン向けの一貫性のあるモデルを分離。システム全体で単一のモデルを共有しない
• Context Mapping: コンテキスト間の関係を定義（Anti-Corruption Layer、Shared Kernel、Open Host Service）
• Ubiquitous Language: コード内のすべての用語はドメイン専門家が使用する用語と一致

戦術的パターン:

• Entities: 時間とともに変化する安定したアイデンティティを持つオブジェクト
• Value Objects: 属性によって識別される不変オブジェクト（Email、Money、Address）
• Aggregates: 一貫性の境界。ルートのみが外部からアクセス可能
• Repositories: 集約を永続化および再構成。ストレージメカニズムを抽象化
• Domain Events: ドメイン内で発生した事象をキャプチャ。集約横断的な調整に使用

Clean Architecture — ディレクトリ構造

app/
├── domain/           # Entities, value objects, interfaces
│   ├── entities/
│   │   ├── user.py
│   │   └── order.py
│   ├── value_objects/
│   │   ├── email.py
│   │   └── money.py
│   └── interfaces/   # Abstract ports (no implementations)
│       ├── user_repository.py
│       └── payment_gateway.py
├── use_cases/        # Application business rules
│   ├── create_user.py
│   ├── process_order.py
│   └── send_notification.py
├── adapters/         # Concrete implementations
│   ├── repositories/
│   │   ├── postgres_user_repository.py
│   │   └── redis_cache_repository.py
│   ├── controllers/
│   │   └── user_controller.py
│   └── gateways/
│       ├── stripe_payment_gateway.py
│       └── sendgrid_email_gateway.py
└── infrastructure/   # Framework wiring, config, DI container
    ├── database.py
    ├── config.py
    └── logging.py

1 文で表した依存規則: domain/ および use_cases/ のすべての import ステートメントは domain/ に向けてのみ指し示す必要があります。これらのレイヤーのどれも adapters/ または infrastructure/ からインポートしてはいけません。

Clean Architecture — コア実装

# domain/entities/user.py
from dataclasses import dataclass
from datetime import datetime

@dataclass
class User:
    """Core user entity — no framework dependencies."""
    id: str
    email: str
    name: str
    created_at: datetime
    is_active: bool = True

    def deactivate(self):
        self.is_active = False

    def can_place_order(self) -> bool:
        return self.is_active


# domain/interfaces/user_repository.py
from abc import ABC, abstractmethod
from typing import Optional
from domain.entities.user import User

class IUserRepository(ABC):
    """Port: defines contract, no implementation details."""

    @abstractmethod
    async def find_by_id(self, user_id: str) -> Optional[User]: ...

    @abstractmethod
    async def find_by_email(self, email: str) -> Optional[User]: ...

    @abstractmethod
    async def save(self, user: User) -> User: ...

    @abstractmethod
    async def delete(self, user_id: str) -> bool: ...


# use_cases/create_user.py
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import uuid
from domain.entities.user import User
from domain.interfaces.user_repository import IUserRepository

@dataclass
class CreateUserRequest:
    email: str
    name: str

@dataclass
class CreateUserResponse:
    user: Optional[User]
    success: bool
    error: Optional[str] = None

class CreateUserUseCase:
    """Use case: orchestrates business logic, no HTTP or DB details."""

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

    async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
        existing = await self.user_repository.find_by_email(request.email)
        if existing:
            return CreateUserResponse(user=None, success=False, error="Email already exists")

        user = User(
            id=str(uuid.uuid4()),
            email=request.email,
            name=request.name,
            created_at=datetime.now(),
        )
        saved_user = await self.user_repository.save(user)
        return CreateUserResponse(user=saved_user, success=True)


# adapters/repositories/postgres_user_repository.py
from domain.interfaces.user_repository import IUserRepository
from domain.entities.user import User
from typing import Optional
import asyncpg

class PostgresUserRepository(IUserRepository):
    """Adapter: PostgreSQL implementation of the user port."""

    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool

    async def find_by_id(self, user_id: str) -> Optional[User]:
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
            return self._to_entity(row) if row else None

    async def find_by_email(self, email: str) -> Optional[User]:
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow("SELECT * FROM users WHERE email = $1", email)
            return self._to_entity(row) if row else None

    async def save(self, user: User) -> User:
        async with self.pool.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO users (id, email, name, created_at, is_active)
                VALUES ($1, $2, $3, $4, $5)
                ON CONFLICT (id) DO UPDATE
                SET email = $2, name = $3, is_active = $5
                """,
                user.id, user.email, user.name, user.created_at, user.is_active,
            )
        return user

    async def delete(self, user_id: str) -> bool:
        async with self.pool.acquire() as conn:
            result = await conn.execute("DELETE FROM users WHERE id = $1", user_id)
            return result == "DELETE 1"

    def _to_entity(self, row) -> User:
        return User(
            id=row["id"], email=row["email"], name=row["name"],
            created_at=row["created_at"], is_active=row["is_active"],
        )


# adapters/controllers/user_controller.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from use_cases.create_user import CreateUserUseCase, CreateUserRequest

router = APIRouter()

class CreateUserDTO(BaseModel):
    email: str
    name: str

@router.post("/users")
async def create_user(
    dto: CreateUserDTO,
    use_case: CreateUserUseCase = Depends(get_create_user_use_case),
):
    """Controller handles HTTP only — no business logic lives here."""
    response = await use_case.execute(CreateUserRequest(email=dto.email, name=dto.name))
    if not response.success:
        raise HTTPException(status_code=400, detail=response.error)
    return {"user": response.user}

Hexagonal Architecture — Ports and Adapters

# Core domain service — no infrastructure dependencies
class OrderService:
    def __init__(
        self,
        order_repository: OrderRepositoryPort,
        payment_gateway: PaymentGatewayPort,
        notification_service: NotificationPort,
    ):
        self.orders = order_repository
        self.payments = payment_gateway
        self.notifications = notification_service

    async def place_order(self, order: Order) -> OrderResult:
        if not order.is_valid():
            return OrderResult(success=False, error="Invalid order")

        payment = await self.payments.charge(amount=order.total, customer=order.customer_id)
        if not payment.success:
            return OrderResult(success=False, error="Payment failed")

        order.mark_as_paid()
        saved_order = await self.orders.save(order)
        await self.notifications.send(
            to=order.customer_email,
            subject="Order confirmed",
            body=f"Order {order.id} confirmed",
        )
        return OrderResult(success=True, order=saved_order)


# Ports (driving and driven interfaces)
class OrderRepositoryPort(ABC):
    @abstractmethod
    async def save(self, order: Order) -> Order: ...

class PaymentGatewayPort(ABC):
    @abstractmethod
    async def charge(self, amount: Money, customer: str) -> PaymentResult: ...

class NotificationPort(ABC):
    @abstractmethod
    async def send(self, to: str, subject: str, body: str): ...


# Production adapter: Stripe
class StripePaymentAdapter(PaymentGatewayPort):
    def __init__(self, api_key: str):
        import stripe
        stripe.api_key = api_key
        self._stripe = stripe

    async def charge(self, amount: Money, customer: str) -> PaymentResult:
        try:
            charge = self._stripe.Charge.create(
                amount=amount.cents, currency=amount.currency, customer=customer
            )
            return PaymentResult(success=True, transaction_id=charge.id)
        except self._stripe.error.CardError as e:
            return PaymentResult(success=False, error=str(e))


# Test adapter: no external dependencies
class MockPaymentAdapter(PaymentGatewayPort):
    async def charge(self, amount: Money, customer: str) -> PaymentResult:
        return PaymentResult(success=True, transaction_id="mock-txn-123")

DDD — Value Objects and Aggregates

# Value Objects: immutable, validated at construction
from dataclasses import dataclass

@dataclass(frozen=True)
class Email:
    value: str

    def __post_init__(self):
        if "@" not in self.value or "." not in self.value.split("@")[-1]:
            raise ValueError(f"Invalid email: {self.value}")

@dataclass(frozen=True)
class Money:
    amount: int   # cents
    currency: str

    def __post_init__(self):
        if self.amount < 0:
            raise ValueError("Money amount cannot be negative")
        if self.currency not in {"USD", "EUR", "GBP"}:
            raise ValueError(f"Unsupported currency: {self.currency}")

    def add(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise ValueError("Currency mismatch")
        return Money(self.amount + other.amount, self.currency)


# Aggregate root: enforces all invariants for its cluster of entities
class Order:
    def __init__(self, id: str, customer_id: str):
        self.id = id
        self.customer_id = customer_id
        self.items: list[OrderItem] = []
        self.status = OrderStatus.PENDING
        self._events: list[DomainEvent] = []

    def add_item(self, product: Product, quantity: int):
        if self.status != OrderStatus.PENDING:
            raise ValueError("Cannot modify a submitted order")
        item = OrderItem(product=product, quantity=quantity)
        self.items.append(item)
        self._events.append(ItemAddedEvent(order_id=self.id, item=item))

    @property
    def total(self) -> Money:
        totals = [item.subtotal() for item in self.items]
        return sum(totals[1:], totals[0]) if totals else Money(0, "USD")

    def submit(self):
        if not self.items:
            raise ValueError("Cannot submit an empty order")
        if self.status != OrderStatus.PENDING:
            raise ValueError("Order already submitted")
        self.status = OrderStatus.SUBMITTED
        self._events.append(OrderSubmittedEvent(order_id=self.id))

    def pop_events(self) -> list[DomainEvent]:
        events, self._events = self._events, []
        return events


# Repository: persist and reconstitute aggregates
class OrderRepository(ABC):
    @abstractmethod
    async def find_by_id(self, order_id: str) -> Optional[Order]: ...

    @abstractmethod
    async def save(self, order: Order) -> None: ...
    # Implementations persist events via pop_events() after writing state

テスト — インメモリアダプター

Clean Architecture が正しく適用されていることを示す最大の特徴は、すべてのユースケースが実行中のデータベースなし、Docker なし、ネットワークなしの単純なユニットテストで実行できることです：

# tests/unit/test_create_user.py
import asyncio
from typing import Dict, Optional
from domain.entities.user import User
from domain.interfaces.user_repository import IUserRepository
from use_cases.create_user import CreateUserUseCase, CreateUserRequest


class InMemoryUserRepository(IUserRepository):
    def __init__(self):
        self._store: Dict[str, User] = {}

    async def find_by_id(self, user_id: str) -> Optional[User]:
        return self._store.get(user_id)

    async def find_by_email(self, email: str) -> Optional[User]:
        return next((u for u in self._store.values() if u.email == email), None)

    async def save(self, user: User) -> User:
        self._store[user.id] = user
        return user

    async def delete(self, user_id: str) -> bool:
        return self._store.pop(user_id, None) is not None


async def test_create_user_succeeds():
    repo = InMemoryUserRepository()
    use_case = CreateUserUseCase(user_repository=repo)

    response = await use_case.execute(CreateUserRequest(email="alice@example.com", name="Alice"))

    assert response.success
    assert response.user.email == "alice@example.com"
    assert response.user.id is not None


async def test_duplicate_email_rejected():
    repo = InMemoryUserRepository()
    use_case = CreateUserUseCase(user_repository=repo)

    await use_case.execute(CreateUserRequest(email="alice@example.com", name="Alice"))
    response = await use_case.execute(CreateUserRequest(email="alice@example.com", name="Alice2"))

    assert not response.success
    assert "already exists" in response.error

トラブルシューティング

ユースケーステストが実行中のデータベースを必要とする

ビジネスロジックがインフラストラクチャレイヤーに漏出しています。すべてのデータベース呼び出しを IRepository インターフェースの背後に移動し、テストでインメモリ実装を注入してください（上記のテストセクションを参照）。ユースケースコンストラクターは具体的なクラスではなく、抽象ポートを受け入れる必要があります。

レイヤー間の循環インポート

一般的な症状は use_cases と adapters 間の ImportError: cannot import name X です。これは、ユースケースが抽象ポートではなく具体的なアダプタークラスをインポートしている場合に発生します。ルールを強制してください：use_cases/ は domain/ からのみインポート（エンティティとインターフェース）。adapters/ または infrastructure/ からのインポートは決してあってはいけません。

フレームワークデコレーターがドメインエンティティに現れる

SQLAlchemy の Column() や Pydantic の Field() アノテーションがドメインエンティティに現れる場合、そのエンティティはもはや純粋ではありません。adapters/repositories/ に別の ORM モデルを作成し、リポジトリの _to_entity() メソッドでドメインエンティティとの間でマッピングしてください。

すべてのロジックがコントローラーで終わっている

コントローラーが HTTP パース応答フォーマット処理を超えて成長する場合、ロジックをユースケースクラスに抽出してください。コントローラーメソッドは 3 つのことだけを実行するべきです：リクエストをパース、ユースケースを呼び出す、レスポンスをマップ。

値オブジェクトがエラーを遅く発生させている

__post_init__（Python）またはコンストラクタで不変性を検証して、無効な Email または Money が構成できないようにしてください。これにより、不正なデータが境界で検出され、ビジネスロジックの深い部分ではなくなります。

Bounded Contexts 間のコンテキストブリード

Order コンテキストが Identity コンテキストから User エンティティをインポートしている場合、Anti-Corruption Layer を導入してください。Order コンテキストは軽量な CustomerId 値オブジェクトのみを保持し、明示的なインターフェースを通じてのみ Identity コンテキストを呼び出す必要があります。

高度なパターン

詳細な DDD Bounded Context マッピング、完全なマルチサービスプロジェクトツリー、Anti-Corruption Layer 実装、Onion Architecture 比較については、以下を参照してください：

• references/advanced-patterns.md

関連スキル

• microservices-patterns — モノリシックをサービスに分解する際に、これらのアーキテクチャパターンを適用する
• cqrs-implementation — Clean Architecture を CQRS コマンド/クエリ分離の構造的基盤として使用
• saga-orchestration — Saga は明確に定義された集約の境界が必要であり、DDD 戦術的パターンがこれを提供する
• event-store-design — 集約によって生成されるドメインイベントはイベントストアに直接フィードされる