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 などを順に実装し、ドメインを一切触れずに機能拡張できることを体験してください。