Hexagonal Architecture（ポート＆アダプター）実践ガイド — クリーンアーキテクチャの核心

1. ドメイン（Domain / Core）
   - ビジネスロジックの核心
   - 外部依存なし（純粋なコード）
   - Entity、Value Object、Domain Service

2. ポート（Port）
   - ドメインと外部世界のインターフェース（契約）
   - Input Port: 外部 → ドメイン（Use Case）
   - Output Port: ドメイン → 外部（Repositoryインターフェース）

3. アダプター（Adapter）
   - ポートの具体的な実装
   - Input Adapter: REST Controller、gRPC Handler、CLI
   - Output Adapter: MySQL Repository、Redis Cache、HTTP Client

# Layered Architecture（従来型）
# Controller → Service → Repository → DB
# 依存方向: 上 → 下（DBに従属）

# Hexagonal Architecture
# Adapter → Port → Domain ← Port ← Adapter
# 依存方向: 外側 → 内側（Domainが中心）

order-service/
├── domain/                    # コアドメイン
│   ├── models/
│   │   ├── order.py          # Entity
│   │   └── order_item.py     # Value Object
│   └── services/
│       └── order_service.py  # Domain Service
├── ports/                     # ポート（インターフェース）
│   ├── input/
│   │   └── order_use_case.py # Input Port
│   └── output/
│       ├── order_repository.py    # Output Port
│       └── payment_gateway.py     # Output Port
├── adapters/                  # アダプター（実装）
│   ├── input/
│   │   ├── rest_controller.py     # REST API
│   │   └── grpc_handler.py        # gRPC
│   └── output/
│       ├── postgres_order_repo.py # PostgreSQL実装
│       ├── redis_order_cache.py   # Redisキャッシュ
│       └── stripe_payment.py      # Stripe決済
└── config/
    └── dependency_injection.py    # DI設定

# domain/models/order.py
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import List
from uuid import UUID, uuid4


class OrderStatus(Enum):
    PENDING = "pending"
    CONFIRMED = "confirmed"
    SHIPPED = "shipped"
    CANCELLED = "cancelled"


@dataclass
class OrderItem:
    product_id: str
    product_name: str
    quantity: int
    unit_price: float

    @property
    def subtotal(self) -> float:
        return self.quantity * self.unit_price


@dataclass
class Order:
    """注文エンティティ — ビジネスルールを含む"""
    id: UUID = field(default_factory=uuid4)
    customer_id: str = ""
    items: List[OrderItem] = field(default_factory=list)
    status: OrderStatus = OrderStatus.PENDING
    created_at: datetime = field(default_factory=datetime.now)

    @property
    def total_amount(self) -> float:
        return sum(item.subtotal for item in self.items)

    def add_item(self, item: OrderItem) -> None:
        if self.status != OrderStatus.PENDING:
            raise ValueError("確定済みの注文には商品を追加できません")
        if item.quantity <= 0:
            raise ValueError("数量は1以上である必要があります")
        self.items.append(item)

    def confirm(self) -> None:
        if not self.items:
            raise ValueError("商品のない注文は確定できません")
        if self.status != OrderStatus.PENDING:
            raise ValueError(f"'{self.status.value}'ステータスでは確定できません")
        self.status = OrderStatus.CONFIRMED

    def cancel(self) -> None:
        if self.status == OrderStatus.SHIPPED:
            raise ValueError("配送済みの注文はキャンセルできません")
        self.status = OrderStatus.CANCELLED

# ports/input/order_use_case.py
from abc import ABC, abstractmethod
from uuid import UUID
from domain.models.order import Order, OrderItem


class CreateOrderUseCase(ABC):
    @abstractmethod
    def execute(self, customer_id: str, items: list[OrderItem]) -> Order:
        pass


class ConfirmOrderUseCase(ABC):
    @abstractmethod
    def execute(self, order_id: UUID) -> Order:
        pass


class CancelOrderUseCase(ABC):
    @abstractmethod
    def execute(self, order_id: UUID) -> Order:
        pass


# ports/output/order_repository.py
from abc import ABC, abstractmethod
from uuid import UUID
from domain.models.order import Order


class OrderRepository(ABC):
    @abstractmethod
    def save(self, order: Order) -> None:
        pass

    @abstractmethod
    def find_by_id(self, order_id: UUID) -> Order | None:
        pass

    @abstractmethod
    def find_by_customer(self, customer_id: str) -> list[Order]:
        pass


# ports/output/payment_gateway.py
from abc import ABC, abstractmethod
from uuid import UUID


class PaymentGateway(ABC):
    @abstractmethod
    def charge(self, order_id: UUID, amount: float, customer_id: str) -> bool:
        pass

    @abstractmethod
    def refund(self, order_id: UUID) -> bool:
        pass

# domain/services/order_service.py
from uuid import UUID
from domain.models.order import Order, OrderItem
from ports.input.order_use_case import (
    CreateOrderUseCase, ConfirmOrderUseCase, CancelOrderUseCase
)
from ports.output.order_repository import OrderRepository
from ports.output.payment_gateway import PaymentGateway


class OrderService(CreateOrderUseCase, ConfirmOrderUseCase, CancelOrderUseCase):
    """注文サービス — Input Portの実装"""

    def __init__(
        self,
        order_repo: OrderRepository,
        payment_gateway: PaymentGateway
    ):
        # Output Portに依存（具体的な実装ではない！）
        self._order_repo = order_repo
        self._payment_gateway = payment_gateway

    def execute(self, customer_id: str = None, items: list[OrderItem] = None,
                order_id: UUID = None) -> Order:
        # dispatch based on params (simplified)
        if customer_id and items:
            return self._create_order(customer_id, items)
        raise ValueError("Invalid parameters")

    def _create_order(self, customer_id: str, items: list[OrderItem]) -> Order:
        order = Order(customer_id=customer_id)
        for item in items:
            order.add_item(item)
        self._order_repo.save(order)
        return order

    def confirm_order(self, order_id: UUID) -> Order:
        order = self._order_repo.find_by_id(order_id)
        if not order:
            raise ValueError(f"注文が見つかりません: {order_id}")

        # 決済処理
        success = self._payment_gateway.charge(
            order_id=order.id,
            amount=order.total_amount,
            customer_id=order.customer_id
        )
        if not success:
            raise ValueError("決済に失敗しました")

        order.confirm()
        self._order_repo.save(order)
        return order

    def cancel_order(self, order_id: UUID) -> Order:
        order = self._order_repo.find_by_id(order_id)
        if not order:
            raise ValueError(f"注文が見つかりません: {order_id}")

        order.cancel()
        self._payment_gateway.refund(order_id)
        self._order_repo.save(order)
        return order

# adapters/output/postgres_order_repo.py
import psycopg2
from uuid import UUID
from domain.models.order import Order, OrderItem, OrderStatus
from ports.output.order_repository import OrderRepository


class PostgresOrderRepository(OrderRepository):
    def __init__(self, connection_string: str):
        self._conn_str = connection_string

    def save(self, order: Order) -> None:
        with psycopg2.connect(self._conn_str) as conn:
            with conn.cursor() as cur:
                cur.execute("""
                    INSERT INTO orders (id, customer_id, status, created_at)
                    VALUES (%s, %s, %s, %s)
                    ON CONFLICT (id) DO UPDATE SET status = %s
                """, (str(order.id), order.customer_id,
                      order.status.value, order.created_at,
                      order.status.value))

                for item in order.items:
                    cur.execute("""
                        INSERT INTO order_items
                        (order_id, product_id, product_name, quantity, unit_price)
                        VALUES (%s, %s, %s, %s, %s)
                        ON CONFLICT DO NOTHING
                    """, (str(order.id), item.product_id,
                          item.product_name, item.quantity, item.unit_price))

    def find_by_id(self, order_id: UUID) -> Order | None:
        with psycopg2.connect(self._conn_str) as conn:
            with conn.cursor() as cur:
                cur.execute("SELECT * FROM orders WHERE id = %s", (str(order_id),))
                row = cur.fetchone()
                if not row:
                    return None
                return self._to_domain(row, cur)

    def find_by_customer(self, customer_id: str) -> list[Order]:
        # 実装省略
        pass

    def _to_domain(self, row, cursor) -> Order:
        # DB行 → ドメインモデル変換
        pass


# adapters/input/rest_controller.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from uuid import UUID
from domain.services.order_service import OrderService
from domain.models.order import OrderItem

app = FastAPI()


class CreateOrderRequest(BaseModel):
    customer_id: str
    items: list[dict]


class OrderResponse(BaseModel):
    id: str
    customer_id: str
    status: str
    total_amount: float


def create_rest_controller(order_service: OrderService):

    @app.post("/orders", response_model=OrderResponse)
    async def create_order(request: CreateOrderRequest):
        items = [
            OrderItem(
                product_id=i["product_id"],
                product_name=i["product_name"],
                quantity=i["quantity"],
                unit_price=i["unit_price"]
            )
            for i in request.items
        ]
        order = order_service._create_order(request.customer_id, items)
        return OrderResponse(
            id=str(order.id),
            customer_id=order.customer_id,
            status=order.status.value,
            total_amount=order.total_amount
        )

    @app.post("/orders/{order_id}/confirm")
    async def confirm_order(order_id: UUID):
        try:
            order = order_service.confirm_order(order_id)
            return {"status": order.status.value}
        except ValueError as e:
            raise HTTPException(status_code=400, detail=str(e))

    return app

# config/dependency_injection.py
from domain.services.order_service import OrderService
from adapters.output.postgres_order_repo import PostgresOrderRepository
from adapters.output.stripe_payment import StripePaymentGateway
from adapters.input.rest_controller import create_rest_controller


def bootstrap():
    # Output Adapters
    order_repo = PostgresOrderRepository(
        connection_string="postgresql://user:pass@localhost/orders"
    )
    payment_gateway = StripePaymentGateway(
        api_key="sk_test_xxx"
    )

    # Domain Service（Port実装）
    order_service = OrderService(
        order_repo=order_repo,
        payment_gateway=payment_gateway
    )

    # Input Adapter
    app = create_rest_controller(order_service)

    return app


# DBを変更したい場合は？
# PostgresOrderRepository → MongoOrderRepositoryに差し替えるだけ！
# ドメインコードの変更は不要！

import pytest
from domain.models.order import Order, OrderItem, OrderStatus


class TestOrder:
    def test_add_item(self):
        order = Order(customer_id="C001")
        item = OrderItem("P001", "ノートPC", 1, 1500000)
        order.add_item(item)
        assert len(order.items) == 1
        assert order.total_amount == 1500000

    def test_cannot_add_item_to_confirmed_order(self):
        order = Order(customer_id="C001")
        order.add_item(OrderItem("P001", "ノートPC", 1, 1500000))
        order.confirm()
        with pytest.raises(ValueError, match="確定済みの注文"):
            order.add_item(OrderItem("P002", "マウス", 1, 50000))

    def test_cannot_confirm_empty_order(self):
        order = Order(customer_id="C001")
        with pytest.raises(ValueError, match="商品のない"):
            order.confirm()

    def test_cannot_cancel_shipped_order(self):
        order = Order(customer_id="C001")
        order.add_item(OrderItem("P001", "ノートPC", 1, 1500000))
        order.confirm()
        order.status = OrderStatus.SHIPPED
        with pytest.raises(ValueError, match="配送済み"):
            order.cancel()

from unittest.mock import MagicMock
from domain.services.order_service import OrderService
from domain.models.order import Order, OrderItem


class TestOrderService:
    def setup_method(self):
        self.mock_repo = MagicMock()
        self.mock_payment = MagicMock()
        self.service = OrderService(self.mock_repo, self.mock_payment)

    def test_create_order(self):
        items = [OrderItem("P001", "ノートPC", 1, 1500000)]
        order = self.service._create_order("C001", items)

        assert order.customer_id == "C001"
        assert len(order.items) == 1
        self.mock_repo.save.assert_called_once()

    def test_confirm_order_with_payment(self):
        order = Order(customer_id="C001")
        order.add_item(OrderItem("P001", "ノートPC", 1, 1500000))
        self.mock_repo.find_by_id.return_value = order
        self.mock_payment.charge.return_value = True

        result = self.service.confirm_order(order.id)

        self.mock_payment.charge.assert_called_once()
        assert result.status.value == "confirmed"