はじめに
「データベースを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が中心）
依存の方向がすべてです
ポートとアダプターという名前のせいで「インターフェースをたくさん作るパターン」と誤解されがちです。しかしこのアーキテクチャが実際に強制するルールは1つ、依存の方向だけです。Alistair Cockburnが原文で示した意図も1文に要約できます。ユーザーが呼ぼうと自動テストが呼ぼうとバッチスクリプトが呼ぼうとアプリケーションが同じように駆動できること、そして最終的に接続されるランタイム機器やデータベースから隔離された状態で開発・テストできること、それだけです。
矢印は2本あります
混乱のほとんどは矢印を1本だと考えるところから始まります。実際には向きが正反対の矢印が2本あります。1本はどのモジュールがどのモジュールを import するかを表すインポートの矢印、もう1本は実行中に誰が誰のメソッドを呼ぶかを表す呼び出しの矢印です。インポートの矢印は常に内側を向き、呼び出しの矢印は外側を向きます。
[インポートの方向] adapters ──import──▶ ports ──import──▶ domain
(外側) (内側)
[呼び出しの方向] domain ──call──▶ ports(抽象型) ──▶ adapters
(内側) (外側)
同じ境界を2本の矢印が互いに逆向きに横切る。
この逆転こそが依存性逆転である。
先ほどのコードで確認しましょう。抽象クラス OrderRepository は ports/output/ の下にあります。ドメイン側のコードです。実装 PostgresOrderRepository は adapters/output/ の下にあり、自分より内側にある OrderRepository を import します。インポートの矢印が外から内へ入ってきたわけです。一方、実行時には OrderService が save() を呼び、その呼び出しは実際にはPostgreSQLアダプターの中で実行されます。呼び出しの矢印は内から外へ出ていきます。この2本が互いに逆向きに置かれた状態、それが依存性逆転です。
ポートを「ドメインが所有するインターフェース」と呼ぶ理由もここにあります。ポートはインフラがドメインに提供するAPIではなく、ドメインがインフラに要求する契約です。だからポートの名前とシグネチャはドメインの語彙で書かれるべきです。find_by_customer(customer_id) はドメインの言葉で、execute_query(sql) はインフラの言葉です。ポートにインフラの言葉が混ざり始めた時点で、それは名前だけのポートになります。
このルールがないと何が壊れるか
依存性逆転を外すと、ドメインコードはたいていこうなります。
# domain/models/order.py ← ドメインなのにインフラをimportしている
from sqlalchemy.orm import Session # ORMセッション型
from infrastructure.db import OrderTable # ORMマッピングクラス
class Order:
def confirm(self, session: Session) -> None:
if not self.items:
raise ValueError("商品のない注文は確定できません")
self.status = "confirmed"
session.merge(OrderTable.from_domain(self)) # ドメインがコミットを知っている
session.commit()
ここで具体的に2つが壊れます。
1つ目、ドメインを単体テストできなくなります。このモジュールを import した瞬間にORMがロードされ、ORMはたいてい設定済みの接続やメタデータを要求します。「商品のない注文は確定できない」という1行のルールを検証するためにテスト用データベースを立ち上げる羽目になります。テストは遅くなり、遅いテストは結局誰も回さなくなります。
2つ目、ライブラリのアップグレードがドメインの変更になります。ORMのメジャーバージョンが上がってセッションAPIが変われば、変更差分の中にドメインファイルが入ってきます。レビュアーはビジネスルールが変わったのか配管が変わったのか区別できません。こうした差分が何度か続くと、ドメインファイルのgit履歴はルールの歴史ではなくライブラリ移行の歴史になります。
つまり「DBを差し替えやすくなる」というよくある宣伝文句は、正確には副次効果です。ほとんどのチームはPostgreSQLからMongoDBへ移ることなくサービスを終えます。本当の利益は、ドメインのルールをインフラ抜きで読み、テストし、直せることのほうにあります。
主導ポートと被主導ポート
Cockburnはポートを2種類に分けます。primary(主導、driving)とsecondary(被主導、driven)です。定義は簡潔で、主導アクターはアプリケーションを駆動する側、被主導アクターはアプリケーションが駆動する側です。
実務で両者を分ける基準は1つで足ります。そのやり取りを誰が先に始めるか。
主導(primary / driving)ポート — 外側が呼び出しを開始する
RESTコントローラー / gRPCハンドラー / CLI / バッチジョブ / キューコンシューマー / 受け入れテスト
│
▼
┌────────────────────────────┐
│ アプリケーション │
│ (ユースケース + ドメイン) │
└────────────────────────────┘
│
▼
被主導(secondary / driven)ポート — アプリケーションが呼び出しを開始する
OrderRepository / PaymentGateway / NotificationSender / Clock
判別法: この会話を誰が先に始めたか?
主導ポート側の実例はこうです。HTTPリクエストを受けるRESTコントローラーはユーザーが始めます。キューからメッセージを取り出して処理するコンシューマーは、ブローカーが押し込んでいるように見えますが、アプリケーションのユースケースを開始する側なので主導です。cronで回る精算バッチや運用者が叩くCLIコマンドも主導です。受け入れテストも同じで、この点は原文が特に強調するところです。テストハーネスはRESTコントローラーとまったく同じ差し込み口に入ります。
被主導ポート側はこうです。OrderRepository は注文を保存し取得します。PaymentGateway は決済を承認し返金します。NotificationSender はメールやプッシュを送ります。ここに時計と乱数生成器を含めるのを忘れがちです。ドメインが現在時刻を直接読むと、そのドメインのテストは実行するたびに違う値を見ることになります。時刻と乱数を被主導ポートに追い出せば、テストは決定的になります。
紛らわしいケースもあります。同じメッセージキューでも、コンシューマー側は主導ポート、プロデューサー側は被主導ポートになります。同じ技術を使っているからといって同じ種類のポートとは限りません。方向を決めるのは技術ではなく呼び出しの起点です。
ポートをいくつ置くかについて原文は慎重です。数を「間違えて」も特に害があるようには見えず、結局は直感の問題だと述べ、著者自身は2つか3つ、多くても4つ程度を好むと書いています。外部システム1つにつきポート1つを機械的に作れというルールは原文のどこにもありません。ポートはアダプターの数ではなく会話の種類で分かれます。
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
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()
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"
要求を1つ最後まで追いかける
ここまでの部品を1つの要求で貫いてみます。要求はこれです。「注文をキャンセルしたら決済を返金し、在庫を戻す」。
まず各部品がどのファイルに落ちるかを決めます。先ほどのプロジェクト構造をそのまま使います。
要求: 注文をキャンセルしたら決済を返金し、在庫を戻す
adapters/input/rest_controller.py DELETE /orders/{id} を受ける ← 主導アダプター
ports/input/order_use_case.py CancelOrderUseCase 契約 ← 主導ポート
domain/services/order_service.py キャンセル手順を調整 ← ユースケース
domain/models/order.py Order.cancel() でルール判定 ← ドメイン
ports/output/order_repository.py save / find_by_id ← 被主導ポート
ports/output/payment_gateway.py refund ← 被主導ポート
ports/output/inventory_port.py restock (今回新規追加) ← 被主導ポート
adapters/output/postgres_order_repo.py 実際のSQL ← 被主導アダプター
adapters/output/stripe_payment.py 実際のStripe呼び出し ← 被主導アダプター
adapters/output/wms_inventory.py 実際のWMS HTTP呼び出し ← 被主導アダプター
新しい要求で増えたのは被主導ポート1つと、そのアダプター1つだけです。在庫を戻すという要求が生まれたからといって、ドメインモデルのキャンセルルールが変わるわけではありません。この区別が保たれているかどうかが、設計が生きているかを見る最初のサインです。
# ports/output/inventory_port.py — ドメインがインフラに要求する契約
from abc import ABC, abstractmethod
from uuid import UUID
class InventoryPort(ABC):
@abstractmethod
def restock(self, order_id: UUID, lines: list[tuple[str, int]]) -> None:
"""(product_id, quantity) の一覧ぶんだけ在庫を戻す。"""
# domain/models/order.py — ルール判定はここだけ
def cancel(self) -> None:
if self.status == OrderStatus.SHIPPED:
raise ValueError("配送済みの注文はキャンセルできません")
if self.status == OrderStatus.CANCELLED:
raise ValueError("すでにキャンセル済みの注文です")
self.status = OrderStatus.CANCELLED
def restock_lines(self) -> list[tuple[str, int]]:
return [(i.product_id, i.quantity) for i in self.items]
# domain/services/order_service.py — 調整だけ、ルールは判定しない
class CancelOrderService(CancelOrderUseCase):
def __init__(self, orders: OrderRepository,
payments: PaymentGateway,
inventory: InventoryPort):
self._orders = orders
self._payments = payments
self._inventory = inventory
def execute(self, order_id: UUID) -> Order:
order = self._orders.find_by_id(order_id)
if order is None:
raise ValueError(f"注文が見つかりません: {order_id}")
order.cancel() # ルールはドメインが判定
self._orders.save(order) # 被主導ポート 1
self._payments.refund(order.id) # 被主導ポート 2
self._inventory.restock(order.id, order.restock_lines()) # 被主導ポート 3
return order
ユースケースに条件分岐が1つもない点に注目してください。「配送済みの注文はキャンセルできない」という判断は Order.cancel() の中にあり、ユースケースは順序を決めてポートを呼ぶ仕事だけをしています。この境界が崩れることが、次節の最初の落とし穴です。
ここからインフラを1つも起動せずに、この要求の全体を検証します。モックライブラリの代わりに手書きの偽アダプターを使うと、何が呼ばれたかがアサーションにそのまま現れるので失敗メッセージが読みやすくなります。
# tests/test_cancel_order.py — 偽アダプターだけで全区間を回す
import pytest
from domain.models.order import Order, OrderItem, OrderStatus
from domain.services.order_service import CancelOrderService
class FakeOrderRepository:
def __init__(self, order):
self._order, self.saved = order, []
def find_by_id(self, order_id):
return self._order
def save(self, order):
self.saved.append(order.status)
class FakePaymentGateway:
def __init__(self):
self.refunded = []
def refund(self, order_id):
self.refunded.append(order_id)
return True
class FakeInventory:
def __init__(self):
self.restocked = []
def restock(self, order_id, lines):
self.restocked.append(lines)
def make_order() -> Order:
order = Order(customer_id="C001")
order.add_item(OrderItem("P001", "ノートパソコン", 1, 1_500_000))
order.add_item(OrderItem("P002", "マウス", 2, 50_000))
return order
def test_cancel_refunds_and_restocks():
order = make_order()
repo, pay, inv = FakeOrderRepository(order), FakePaymentGateway(), FakeInventory()
service = CancelOrderService(repo, pay, inv)
result = service.execute(order.id)
assert result.status is OrderStatus.CANCELLED
assert repo.saved == [OrderStatus.CANCELLED]
assert pay.refunded == [order.id]
assert inv.restocked == [[("P001", 1), ("P002", 2)]]
def test_shipped_order_is_not_cancellable():
order = make_order()
order.status = OrderStatus.SHIPPED
service = CancelOrderService(FakeOrderRepository(order),
FakePaymentGateway(), FakeInventory())
with pytest.raises(ValueError, match="配送済み"):
service.execute(order.id)
期待される出力です。以下はpytest 8系での出力形式で、2回目の実行はユースケースから返金呼び出しをわざと削ったあとに回したものです。
$ pytest tests/test_cancel_order.py -q
.. [100%]
2 passed in 0.03s
$ pytest tests/test_cancel_order.py -q # refund呼び出しを削ったあと
.F [100%]
=================================== FAILURES ===================================
_____________________ test_cancel_refunds_and_restocks ________________________
E AssertionError: assert [] == [UUID('9f0c1a3e-...')]
E Right contains one more item: UUID('9f0c1a3e-...')
tests/test_cancel_order.py:57: AssertionError
1 failed, 1 passed in 0.04s
注目すべきは実行時間です。データベースも決済会社もWMSもなしに、ミリ秒で終わります。これほど速いテストだけがコミットのたびに回り、コミットのたびに回るテストだけが実際にリグレッションを止めます。ヘキサゴナルが与える利益の大半は、実はこの地点で回収されます。
落とし穴と失敗事例
落とし穴1. 貧血ドメイン
まず症状から。ユースケースのクラスが伸び続けます。OrderService が300行を超え、スプリントごとに条件分岐が1つずつ増えます。一方で Order エンティティは何か月も変わらず、開いてみるとgetterとsetterしかありません。クラス図はドメインモデルの形をしているのに、実際のルールはすべてサービス側にあります。
診断はこの順で行います。1つ目、ドメインエンティティが例外を投げている箇所を探します。ルール違反を知らせる例外がエンティティに1つもなければ、ルールがエンティティにないということです。2つ目、ユースケースがエンティティの状態フィールドを読んで分岐している行を数えます。状態を見て分岐するコードが複数のユースケースに散っていれば、その判定はもともとエンティティの仕事です。3つ目、同じルールが2つ以上のユースケースに重複していないか確認します。重複が見つかれば診断は確定です。
# ❌ 貧血: 判定がユースケースにある
class OrderService:
def cancel(self, order_id):
order = self._orders.find_by_id(order_id)
if order.status == "shipped": # ルールがここにある
raise ValueError("配送済みの注文はキャンセル不可")
if order.status == "cancelled": # 別のユースケースにも同じものがある
raise ValueError("すでにキャンセル済み")
order.status = "cancelled" # 状態を外から書き換えている
self._orders.save(order)
# ✅ 判定をエンティティへ移す
class OrderService:
def cancel(self, order_id):
order = self._orders.find_by_id(order_id)
order.cancel() # ルールはエンティティが判定
self._orders.save(order)
処方は、状態を外から代入しているコードを探してエンティティのメソッドへ移すことです。order.status = ... のような代入がドメインの外に残っている限り、ルールはいつでもまた漏れ出します。ルールがエンティティに集まれば、ユースケースは自然と短くなります。
落とし穴2. 漏れるポート
症状はアダプターを差し替えようとした瞬間に現れます。インメモリの偽実装を作ろうとすると、ポートの戻り値の型がORMモデルなので真似のしようがありません。あるいはポートのメソッドがセッションオブジェクトを引数に取っていて、セッションのない実装ではそもそもシグネチャを満たせません。ポートがインフラの型を露出した瞬間、そのポートは契約ではなくORMの別名になります。
# ❌ 漏れるポート — インフラがシグネチャに現れている
class OrderRepository(ABC):
@abstractmethod
def find_by_id(self, session, order_id) -> "OrderTable": # ORMモデルを返す
...
@abstractmethod
def execute_query(self, sql: str) -> list[tuple]: # ストレージの語彙
...
# ✅ ドメインの型とドメインの語彙だけを残す
class OrderRepository(ABC):
@abstractmethod
def find_by_id(self, order_id: UUID) -> Order | None:
...
@abstractmethod
def find_by_customer(self, customer_id: str) -> list[Order]:
...
診断はgrep1行で済むことが多いです。ドメインパッケージとポートパッケージからインフラのimportを探し、1行でも出てくればその時点ですでに漏れています。
# ドメイン/ポートがインフラをimportしていないか — 結果が空なら正常
grep -rnE '^[[:space:]]*(from|import)[[:space:]]+(sqlalchemy|psycopg2|pymongo|django|fastapi|redis|boto3)' \
domain/ ports/
# CIにそのまま入れられる形 (1つでも引っかかれば失敗)
if grep -rqE '^[[:space:]]*(from|import)[[:space:]]+(sqlalchemy|psycopg2|django|fastapi)' domain/ ports/; then
echo "FAIL: ドメインまたはポートがインフラをimportしています"
exit 1
fi
この検査をCIに入れておけば、ルールが人の記憶ではなくパイプラインに残ります。自動で検査されないアーキテクチャのルールは、数か月のうちにたいてい崩れます。
落とし穴3. DTOの爆発
境界ごとに型を新しく作っていくと、同じ注文1つを表すクラスが4つも5つもできます。リクエストDTO、ドメインエンティティ、永続化モデル、レスポンスDTO、そして外部の決済会社に送るペイロードまで。フィールドを1つ追加すると5か所を直すことになり、マッピングコードがドメインコードより長くなります。
このコストは実在し、無条件に受け入れるべきものでもありません。判断基準を2つに絞るとだいたい合います。1つ目、2つの型は違う理由で変わるか。APIレスポンスのスキーマはクライアントの都合で変わり、ドメインモデルはルールのために変わります。変わる理由が違うなら分ける価値があります。2つ目、その境界を越える型は外部に公開されるか。公開APIのレスポンス型にドメインエンティティをそのまま使うと、ドメインのフィールド名を変えた瞬間にそれがAPIの破壊的変更になります。
逆に、内部でしか使われずドメインと同じ理由で一緒に変わる層なら、マッピングを省いてかまいません。永続化モデルとドメインエンティティが事実上同じ形で、今後も一緒に変わることが明らかなら、2組を維持するのは原則のための原則です。この判断は毎回やり直す必要があり、答えが1つに決まっているわけではありません。
落とし穴4. トランザクション境界
実務でいちばん難しい問題です。ドメインはトランザクションを知らないべきなのに、先ほどのキャンセルのユースケースでは注文の保存と在庫の復元が一緒にコミットされる必要があります。片方だけ成功すると、在庫が増えたまま注文が生きているか、その逆になります。
正直に言えば万能の解はなく、よく使われる答えが3つあり、それぞれにコストがあります。
1つ目、トランザクションスクリプト方式です。ユースケースの外側、つまりインバウンドアダプターやデコレーターでトランザクションを開いて閉じます。ドメインは完全にきれいなままです。コストは、トランザクション境界がユースケース境界とずれやすいことです。1つのリクエストで2つのユースケースを呼ぶと意図せず1つのトランザクションになり、その事実がコードのどこにも現れません。
2つ目、Unit of Workをポートにする方式です。トランザクションというインフラの概念の代わりに「1つの作業単位」というドメインの概念をポートとして露出し、コミットが実際に何なのかはアダプターだけが知ります。コストは、この抽象が漏れやすいことです。分離レベル、ネストしたトランザクション、セーブポイントなどが必要になった瞬間、ポートのシグネチャがデータベースに似てきます。
# ports/output/unit_of_work.py — ドメインの語彙で包んだトランザクション境界
from abc import ABC, abstractmethod
class UnitOfWork(ABC):
"""1つの作業単位という概念だけを露出する。
コミットとロールバックが実際に何かはアダプターだけが知る。"""
@abstractmethod
def __enter__(self) -> "UnitOfWork":
...
@abstractmethod
def __exit__(self, exc_type, exc, tb) -> None:
...
@abstractmethod
def commit(self) -> None:
...
# ユースケース側
def execute(self, order_id: UUID) -> Order:
with self._uow:
order = self._orders.find_by_id(order_id)
order.cancel()
self._orders.save(order)
self._inventory.restock(order.id, order.restock_lines())
self._uow.commit()
# 返金は同じトランザクションに入れられない (外部システム)
# → アウトボックスに記録し、別のワーカーが再試行する
self._outbox.append("order.cancelled", order.id)
return order
3つ目、結果整合性を受け入れる方式です。上のコードの最後の行がそれです。同じデータベースにあるものだけを1つのトランザクションにまとめ、外部システムの呼び出しはアウトボックステーブルに記録して別のワーカーが再試行します。コストは明確で、再試行がある以上、受け側は冪等でなければならず、「キャンセルは済んだが返金はまだ」という中間状態が実際に存在するので、運用とサポートがその状態を理解できる必要があります。
どれを選んでもタダではありません。ただし決済会社やWMSのような外部システムはそもそも自社のデータベーストランザクションに参加できないので、2つ以上の外部システムが絡んだ時点で3つ目が事実上唯一の現実的な選択肢になります。
使わないほうがよい場合
このアーキテクチャを使わないほうがよい場面ははっきりあります。
1つ目、ドメインルールのないCRUDサービスです。リクエストを受けて検証し、1つのテーブルに入れて返すだけなら、ポートとアダプターは安全性ではなく移動距離だけを増やします。フィールドを1つ足すのにリクエストDTO、ドメインモデル、ポートのシグネチャ、アダプターのマッピング、レスポンスDTOを順に直すことになります。こういうサービスではコントローラーがORMを直接使うほうが読みやすく直しやすいです。
2つ目、アダプターを結局差し替えない小さなチームです。間接層のコストは毎日支払われますが、利益は差し替えかテストの時点でしか回収されません。回収の時が来なければコストだけが残ります。
3つ目、寿命の短いコードです。実験用のサービスや1四半期だけ使って捨てるツールには、フレームワークより長く生き残るドメインがありません。このアーキテクチャの前提そのものが成り立ちません。
そこで導入の前提を2つに整理できます。1つは、フレームワークより長く生き残る本物のドメインロジックがあるか。「状態遷移のルール」「料金計算のルール」「承認条件」のように言葉にできるルールが複数あり、それが頻繁に変わるなら前提は成り立ちます。もう1つは、ポートごとにアダプターが2つ以上あるか。ここでテストダブルを2つ目のアダプターとして数えるのは正当で、原文自身もモックアダプターでアプリケーションを完全に隔離して動かせることを最大の利益として挙げています。ただし、そのテストが実際に存在してはじめて数えられます。誰も2度目の実装をしないインターフェースは2つ目の実装ではなく、ただのファイルです。
最後に、これは全か無かではありません。決済や精算のようにルールが集まっているモジュールにだけポートとアダプターを適用し、残りは普通の階層型にしておく折衷が実務ではもっとも一般的で、たいていうまく機能します。
まとめ
Hexagonal Architectureの核心的な価値：
- ドメインの独立性: ビジネスロジックがDB・フレームワークに依存しない
- 交換の容易さ: アダプターを差し替えるだけで外部システムを変更可能
- テストの容易さ: ドメインはMockなしで、サービスはポートのMockでテスト
- ポートが契約: インターフェース（Port）が内部と外部の明確な契約の役割を果たす
参考資料
- Alistair Cockburn, "Hexagonal Architecture"（Ports and Adaptersの原文）— 1文の意図、主導アクターと被主導アクターの定義、ポートの数についての著者の立場、モックアダプターでアプリケーションを隔離して動かす利益は、すべてここから取っています。https://alistair.cockburn.us/hexagonal-architecture/ （2026-08-16 確認）
この記事のコードはPython 3.10以降の構文（X | None 形式のユニオン型）を前提としており、テスト出力の例はpytest 8系に基づいています。バージョンによって出力形式は異なるため、実際の値はご自身の環境で確認してください。
クイズ（6問）
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での実践実装、テスト戦略まで。ビジネスロジックを外部依存から完全に分離する方法を解説します。
Q2: 核心概念とは何ですか？
3つの核心コンポーネント Layered Architectureとの違い
Q3: Python実践実装の核心的な概念を説明してください。
プロジェクト構造 ドメインモデル ポート定義 ドメインサービス（Use Case実装） アダプター実装
依存性注入の設定
Q4: テスト戦略の主な特徴は何ですか？
ドメイン単体テスト（外部依存なし） Mockを使ったサービステスト
Q5: インポートの矢印と呼び出しの矢印はそれぞれどちらを向きますか？
インポート（コンパイル時）の矢印はアダプターからドメインへ、つまり常に内側を向きます。呼び出し（実行時）の矢印はドメインからポートを経てアダプターへ、つまり外側を向きます。同じ境界を2本の矢印が互いに逆向きに横切る状態が依存性逆転です。
Q6: 主導ポートと被主導ポートを実務でどう見分けますか？
基準は1つ、そのやり取りを誰が先に始めるかです。外側がアプリケーションを呼べば主導（RESTコントローラー、CLI、バッチジョブ、キューコンシューマー、受け入れテスト）、アプリケーションが外を呼べば被主導（リポジトリ、決済ゲートウェイ、通知送信、時計）です。同じメッセージキューでもコンシューマー側は主導、プロデューサー側は被主導になります。
Q7: 貧血ドメインの症状と診断の順序は何ですか？
症状はユースケースのクラスだけが伸び続け、エンティティにはgetterとsetterしかない状態です。診断はエンティティにルール違反の例外があるか確認し、ユースケースがエンティティの状態を読んで分岐している行を数え、同じルールが2つのユースケースに重複していないか確認する順で行います。
Q8: 2つのアウトバウンドアダプターが一緒にコミットされるべきとき、どんな選択肢がありますか？
ユースケースの外でトランザクションを開く方式は境界がユースケースとずれやすく、Unit of Workポートは分離レベルやセーブポイントが必要になると抽象が漏れます。アウトボックスによる結果整合性は受け側の冪等性と中間状態の運用負担を要求します。外部システムは自社のトランザクションに参加できないため、3つ目が現実的な場合が多いです。
Q9: ヘキサゴナルアーキテクチャを使わないほうがよいのはどんな場合ですか？
ドメインルールのないCRUDサービス、アダプターを結局差し替えない小さなチーム、寿命の短い実験用コードです。導入の前提はフレームワークより長く生き残るドメインロジックがあることと、ポートごとにアダプターが2つ以上あることで、実際に動くテストダブルは2つ目として数えてかまいません。