architecture-patterns

アーキテクチャパターン

Clean Architecture、Hexagonal Architecture、Domain-Driven Design などの実績あるバックエンドアーキテクチャパターンを習得し、保守性、テスト容易性、スケーラビリティに優れたシステムを構築します。

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

• 新しいバックエンドシステムを一から設計するとき
• 保守性を高めるためにモノリシックアプリケーションをリファクタリングするとき
• チームのアーキテクチャ標準を定めるとき
• 密結合なアーキテクチャから疎結合なアーキテクチャへ移行するとき
• ドメイン駆動設計の原則を実装するとき
• テストやモックが容易なコードベースを作成するとき
• マイクロサービスへの分割を計画するとき

主な概念

1. Clean Architecture（Uncle Bob）

レイヤー（依存関係は内側に向かいます）：

• エンティティ：中核となるビジネスモデル
• ユースケース：アプリケーションのビジネスルール
• インターフェースアダプター：コントローラー、プレゼンター、ゲートウェイ
• フレームワークとドライバー：UI、データベース、外部サービス

主な原則：

• 依存関係は内側を向きます
• 内側のレイヤーは外側のレイヤーについて何も知りません
• ビジネスロジックはフレームワークから独立しています
• UI、データベース、外部サービスなしでテストできます

2. Hexagonal Architecture（ポートとアダプター）

構成要素：

• ドメインコア：ビジネスロジック
• ポート：やり取りを定義するインターフェース
• アダプター：ポートの実装（データベース、REST、メッセージキュー）

メリット：

• 実装を簡単に切り替えられます（テスト用のモックなど）
• テクノロジーに依存しないコア
• 関心事を明確に分離できます

3. Domain-Driven Design（DDD）

戦略的パターン：

• 境界づけられたコンテキスト：異なるドメインごとにモデルを分けます
• コンテキストマッピング：コンテキスト間の関係を定義します
• ユビキタス言語：共有する用語体系を定めます

戦術的パターン：

• エンティティ：識別子を持つオブジェクト
• 値オブジェクト：属性によって定義される不変オブジェクト
• 集約：整合性の境界
• リポジトリ：データアクセスの抽象化
• ドメインイベント：発生した出来事

Clean Architecture パターン

ディレクトリ構造

app/
├── domain/           # Entities & business rules
│   ├── entities/
│   ├── value_objects/
│   └── interfaces/   # Abstract interfaces
├── use_cases/        # Application business rules
├── adapters/         # Interface implementations
│   ├── repositories/
│   ├── controllers/
│   └── gateways/
└── infrastructure/   # Framework & external concerns

実装例

# 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):
        """Business rule: deactivating user."""
        self.is_active = False

    def can_place_order(self) -> bool:
        """Business rule: active users can order."""
        return self.is_active

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

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

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

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

# use_cases/create_user.py
from dataclasses import dataclass
import uuid
from datetime import datetime

@dataclass
class CreateUserRequest:
    email: str
    name: str

class CreateUserUseCase:
    """Use case: orchestrates business logic."""

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

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

        # Create entity
        user = User(
            id=str(uuid.uuid4()),
            email=request.email,
            name=request.name,
            created_at=datetime.now(),
            is_active=True
        )

        # Persist
        saved_user = await self.user_repository.save(user)
        return CreateUserResponse(user=saved_user, success=True)

# adapters/repositories/postgres_user_repository.py
class PostgresUserRepository(IUserRepository):
    """Adapter: PostgreSQL implementation."""

    def __init__(self, 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 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

Hexagonal Architecture パターン

# Core domain (hexagon center)
class OrderService:
    """Domain service - no infrastructure dependencies."""

    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:
        # Business logic
        if not order.is_valid():
            return OrderResult(success=False, error="Invalid order")

        # Use ports (interfaces)
        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)

# Adapters
class StripePaymentAdapter(PaymentGatewayPort):
    """Primary adapter: connects to Stripe API."""
    async def charge(self, amount: Money, customer: str) -> PaymentResult:
        # Implementation

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

Domain-Driven Design パターン

# Value Objects (immutable)
@dataclass(frozen=True)
class Email:
    value: str

    def __post_init__(self):
        if "@" not in self.value:
            raise ValueError("Invalid email")

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

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

# Entities (with identity)
class Order:
    def __init__(self, id: str, customer: Customer):
        self.id = id
        self.customer = customer
        self.items: List[OrderItem] = []
        self.status = OrderStatus.PENDING
        self._events: List[DomainEvent] = []

    def add_item(self, product: Product, quantity: int):
        item = OrderItem(product, quantity)
        self.items.append(item)
        self._events.append(ItemAddedEvent(self.id, item))

    def submit(self):
        if not self.items:
            raise ValueError("Cannot submit empty order")
        self.status = OrderStatus.SUBMITTED
        self._events.append(OrderSubmittedEvent(self.id))

# Aggregates (consistency boundary)
class Customer:
    def __init__(self, id: str, email: Email):
        self.id = id
        self.email = email
        self._addresses: List[Address] = []

    def add_address(self, address: Address):
        if len(self._addresses) >= 5:
            raise ValueError("Maximum 5 addresses allowed")
        self._addresses.append(address)

ベストプラクティス

1. 依存関係のルール：依存関係は常に内側を向けます
2. インターフェース分離：小さく、目的を絞ったインターフェースを使います
3. ドメイン内のビジネスロジック：フレームワークをコアから切り離します
4. テストの独立性：コアはインフラストラクチャなしでテストできます
5. 境界づけられたコンテキスト：ドメインの境界を明確にします
6. ユビキタス言語：一貫した用語を使います
7. 薄いコントローラー：処理をユースケースに委譲します
8. リッチドメインモデル：データとともに振る舞いを持たせます

よくある落とし穴

• 貧血ドメイン：データだけで、振る舞いを持たないエンティティ
• フレームワークへの結合：ビジネスロジックがフレームワークに依存している
• 肥大化したコントローラー：コントローラーにビジネスロジックがある
• リポジトリからの漏洩：ORM オブジェクトを公開する
• 抽象化の不足：コアに具象的な依存関係がある
• 過剰設計：単純な CRUD にクリーンアーキテクチャを適用する