#0116（2025/05/02）Port-Adapterパターンとは

# 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

```python
# domain/models.py
from dataclasses import dataclass
@dataclass
class User:
    name: str
    email: str
```

```python
# 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

```python
# 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

```python
# 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()
```

```python
# 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}")
```

```python
# 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

```python
# 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

```python
# 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

```python
# 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. よくある誤解とアンチパターン

1. **Adapter にビジネスロジックを書く** → ドメインから漏れるので NG。
2. **Port を具象 class にしてしまう** → 抽象(ABC/Protocol)でなければ差替不能。
3. **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 などを順に実装し、ドメインを一切触れずに機能拡張できることを体験してください。

---