Port‑Adapterパターンについて
1. はじめに
モノリス／マイクロサービス、Web／CLI／IoT—私たちは多様な入出力やストレージを抱えながらソフトウェアを育て続けます。そんな“つぎはぎ”を整理し、ビジネスロジックを変更に強く保つためのアーキテクチャが Port & Adapter（Hexagonal Architecture） です。本記事では、由来・概念から Python & FastAPI 実装、テスト、運用時の変更シナリオまで、実務で活かせるレベルで徹底解説します。
2. 命名の由来と六角形の意味
- Alistair Cockburn が 2005 年に発表。
- 図示するとき 六角形 を使い、各辺に「ポート」を均等配置して“どの方向からも接続可能”を示す —— 形状自体に意味はなく、メタファー。
- Clean Architecture, Onion Architecture は同じ依存制御ルールを円筒や層で表した派生形。
+-----------+
UI | Adapter |
------+-----------+------ 他の入出力
| ▲ |
| Port |
+----|------+
▼
+---------+
| Domain |
+---------+
▲
+----|------+
| Port |
------+-----------+------ DB／External API
| Adapter |
+-----------+
3. 主要コンポーネント
| レイヤ | 役割 | 実装例 |
|---|---|---|
| Domain (Core) | ビジネスルール／エンティティ。外部に依存しない |
User dataclass, 集約, ドメインサービス |
| Port | 抽象インターフェース。Domain→外部(出力ポート)／外部→Domain(入力ポート) を分離 |
UserRepository, NotificationService
|
| Adapter | Port を実装し実際の IO を行う |
SQLiteUserRepository, SendGridNotifier, FastAPI Router |
| Application Service | ユースケース単位の調整役。複数 Port を組み合わせる | UserService.register_user() |
| Composition Root | Adapters を組み立てアプリを起動 | main.py |
4. Python / FastAPI サンプル実装
4.1 ディレクトリ
project/
├── domain/
│ ├── models.py
│ └── ports.py
├── application/
│ └── user_service.py
├── adapters/
│ ├── sqlite_repository.py
│ ├── email_notifier.py
│ └── fastapi_controller.py
├── main.py
└── tests/
├── test_user_service.py
└── test_fastapi_app.py
4.2 Domain & Ports
# domain/models.py
from dataclasses import dataclass
@dataclass
class User:
name: str
email: str
# domain/ports.py
from abc import ABC, abstractmethod
from domain.models import User
class UserRepository(ABC):
@abstractmethod
def save(self, user: User): ...
class NotificationService(ABC):
@abstractmethod
def send_welcome_email(self, user: User): ...
4.3 Application Service
# application/user_service.py
from domain.models import User
from domain.ports import UserRepository, NotificationService
class UserService:
def __init__(self, repo: UserRepository, notifier: NotificationService):
self.repo = repo
self.notifier = notifier
def register_user(self, name: str, email: str):
user = User(name, email)
self.repo.save(user)
self.notifier.send_welcome_email(user)
return user
4.4 Adapters
# adapters/sqlite_repository.py
import sqlite3
from domain.ports import UserRepository
from domain.models import User
class SQLiteUserRepository(UserRepository):
def __init__(self, db_path="users.db"):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self.conn.execute("CREATE TABLE IF NOT EXISTS users (name TEXT, email TEXT)")
def save(self, user: User):
self.conn.execute("INSERT INTO users (name, email) VALUES (?, ?)", (user.name, user.email))
self.conn.commit()
# adapters/email_notifier.py
from domain.ports import NotificationService
from domain.models import User
class ConsoleEmailNotifier(NotificationService):
def send_welcome_email(self, user: User):
print(f"[Email] Welcome {user.name}! → {user.email}")
# adapters/fastapi_controller.py
from fastapi import APIRouter
from pydantic import BaseModel
from application.user_service import UserService
class UserRequest(BaseModel):
name: str
email: str
def create_user_router(service: UserService):
router = APIRouter(prefix="/users")
@router.post("")
def register(req: UserRequest):
user = service.register_user(req.name, req.email)
return {"message": f"User {user.name} registered"}
return router
4.5 Composition Root
# main.py
from fastapi import FastAPI
from adapters.sqlite_repository import SQLiteUserRepository
from adapters.email_notifier import ConsoleEmailNotifier
from adapters.fastapi_controller import create_user_router
from application.user_service import UserService
app = FastAPI()
repo = SQLiteUserRepository()
notifier = ConsoleEmailNotifier()
service = UserService(repo, notifier)
app.include_router(create_user_router(service))
5. テスト戦略（pytest）
5.1 単体テスト – Application Layer
# tests/test_user_service.py
import pytest
from application.user_service import UserService
from domain.models import User
class FakeRepo:
def __init__(self): self.saved = []
def save(self, user: User): self.saved.append(user)
class FakeNotifier:
def __init__(self): self.sent = []
def send_welcome_email(self, user: User): self.sent.append(user)
def test_register():
svc = UserService(FakeRepo(), FakeNotifier())
u = svc.register_user("Keita", "k@example.com")
assert u.email == "k@example.com"
5.2 結合テスト – FastAPI + Service
# tests/test_fastapi_app.py
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_create_user():
r = client.post("/users", json={"name":"A","email":"a@x.com"})
assert r.status_code==200
6. 変更シナリオと影響範囲
| 変更要求 | 影響 | 対応方法 |
|---|---|---|
| DB を PostgreSQL に | Output Adapter のみ |
PostgreSQLUserRepository を実装し注入 |
| 通知を SendGrid メールへ | Output Adapter のみ |
SendGridNotifier 実装 |
| UI に CLI を追加 | Input Adapter 追加 |
click or Typer で CLI → UserService 呼出 |
| バッチ処理で定期登録 | Input Adapter 追加 |
APScheduler などでスケジューラ Adapter |
| API v2 でレスポンス変更 | FastAPI Router だけ | ルーター層でバージョニング |
7. よくある誤解とアンチパターン
- Adapter にビジネスロジックを書く → ドメインから漏れるので NG。
- Port を具象 class にしてしまう → 抽象(ABC/Protocol)でなければ差替不能。
- Adapter が一つしか無いのに分ける意味？ → 将来へ投資。ユニットテスト・変更耐性を優先。
8. Clean / Onion との比較
| 観点 | Hexagonal | Clean Architecture | Onion |
|---|---|---|---|
| 可視化図形 | 六角形 | 同心円＋層 | 同心円 |
| 入力/出力ポート | 明示的 | UseCase / Interface Boundary | Interface 層 |
| “外部”の定義 | UI・DBとも辺に置く | 最外層（Framework） | Infrastructure ring |
いずれも依存の方向が内側→外側（ビジネスルールは外部に依存しない）という原則は共通です。
9. まとめ
Port‑Adapter (Hexagonal) は、
- ドメインを外部から隔離し、
- 入出力の変更をアダプター差替えで吸収し、
-
テストと保守のコストを下げる
設計手法です。Python でも ABC と DI をうまく使えば簡潔に実現できます。まずは既存コードの「外部依存」部分を洗い出し、Port 抽象を立て、Adapter を実装してみましょう。
Next Step: SendGrid 通知 / PostgreSQL Adapter / CLI Adapter などを順に実装し、ドメインを一切触れずに機能拡張できることを体験してください。