はじめに
「データベースをMySQLからPostgreSQLに変更する必要があります。」
「REST APIの代わりにgRPCもサポートする必要があります。」
このような要件を聞いたとき、コード全体を書き直す必要があるなら、アーキテクチャに問題があります。**Hexagonal Architecture（ヘキサゴナルアーキテクチャ）**は、ビジネスロジックを外部依存から完全に分離し、このような変更を容易にします。
核心概念
3つの核心コンポーネント
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との違い
Layered Architecture（従来型）
Controller → Service → Repository → DB
依存方向: 上 → 下（DBに従属）
Hexagonal Architecture
Adapter → Port → Domain ← Port ← Adapter
依存方向: 外側 → 内側（Domainが中心）
Python実践実装
プロジェクト構造
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
ドメインサービス（Use Case実装）
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
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に差し替えるだけ！
ドメインコードの変更は不要！
テスト戦略
ドメイン単体テスト（外部依存なし）
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()
Mockを使ったサービステスト
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"
まとめ
Hexagonal Architectureの核心的な価値：
1. **ドメインの独立性**: ビジネスロジックがDB・フレームワークに依存しない
2. **交換の容易さ**: アダプターを差し替えるだけで外部システムを変更可能
3. **テストの容易さ**: ドメインはMockなしで、サービスはポートのMockでテスト
4. **ポートが契約**: インターフェース（Port）が内部と外部の明確な契約の役割を果たす
**Q1. Hexagonal Architectureの3つの核心コンポーネントは？**
Domain（Core）、Port（インターフェース）、Adapter（実装）
**Q2. Input PortとOutput Portの違いは？**
Input Port: 外部からドメインへ（Use Case）。Output Port: ドメインから外部へ（Repositoryインターフェース）
**Q3. Hexagonal Architectureにおける依存方向は？**
外側（Adapter）→ 内側（Domain）。ドメインは外部に依存しない
**Q4. DBをMySQLからPostgreSQLに変更する際に修正が必要なものは？**
Output Adapterのみ差し替え（ドメインコードの変更は不要）
**Q5. ドメイン単体テストでMockが不要な理由は？**
ドメインモデルに外部依存がないため、純粋なロジックのみテスト可能
**Q6. Layered Architectureに対するHexagonalの主な利点は？**
ビジネスロジックがDB/フレームワークに依存しないため、交換とテストが容易
クイズ
Q1: 「Hexagonal Architecture（ポート＆アダプター）実践ガイド —
クリーンアーキテクチャの核心」の主なトピックは何ですか？
Hexagonal Architecture（Ports & Adapters）の核心概念からPython/Spring
Bootでの実践実装、テスト戦略まで。ビジネスロジックを外部依存から完全に分離する方法を解説します。
3つの核心コンポーネント Layered Architectureとの違い
プロジェクト構造 ドメインモデル ポート定義 ドメインサービス（Use Case実装） アダプター実装
依存性注入の設定
ドメイン単体テスト（外部依存なし） Mockを使ったサービステスト
현재 단락 (1/336)
「データベースをMySQLからPostgreSQLに変更する必要があります。」