マイグレーション戦略

既存のサービスクラスや ORM 中心のコードを一度に書き換えると、境界・永続化・観測性の穴が同時に広がる。Kamae では触れたワークフローごとに、パース → 型付き状態 → ポート分離 → 原子性永続化の順で段階的に締める。

最初の境界改善は境界防御、マッパー導入は ORM アダプター、コマンド単位の一貫性は永続化、集約、イベントを参照する。

デフォルトは Strangler Fig パターンである。新フローと高リスク領域を先に移行し、触れない安定レガシー経路はそのままにする。

原則

製品作業を全面マイグレーションでブロックしない。
ビジネスルールを書き換える前に境界を改善する。
アグリゲートまたはワークフローを 1 つずつ移行する。
ユーザーが依存する振る舞いにテストを置く。
検証を弱めない限り、リポジトリの既存規約に従う。

段階的ロールアウト

フェーズ	目標	典型的なタッチポイント	リスク
0 — ベースライン	uv、Ruff、mypy、触れたコードへの pytest	`pyproject.toml`、CI	低
1 — 境界パース	エッジで Pydantic により外部データを検証	API DTO、DB 行モデル、キューペイロード	低
2 — 状態形状	新規/変更フローで `status + Optional[...]` blob を判別共用体に置換	1 ワークフローのドメインモデル	中
3 — 純粋遷移	サービスメソッドからルールを名前付き関数へ	application/domain モジュール	中
4 — ポート/アダプター	ORM/SDK を `Protocol` ポートの背後に隠す	リポジトリ、クライアント	中〜高
5 — 原子性永続化	状態 + イベントを一緒に保存、冪等性/バージョニング追加	リポジトリアダプター、アウトボックス	高
6 — 厳格ゲート	mypy カバレッジ拡大、移行パスでポリシーチェッカー有効化	CI、`files` 設定	継続

全体で次のフェーズに進む前に、別ワークフローで先に着手してよい。ただし 1 ワークフロー内ではフェーズ順序を守る。

フェーズ詳細

フェーズ 1: まず境界パース

リスクが最も低い改善である。最初の段階では内部のドメインコードはそのままにし、エッジで未検証データを止める。

RequestRowAdapter = TypeAdapter(RequestRow)

def row_to_waiting(row: Mapping[str, object]) -> Waiting:
    dto = RequestRowAdapter.validate_python(row)
    return Waiting(...)

フェーズ 2: レガシーモデルの横に判別共用体を導入

初日に古い TaxiRequestService クラスを削除しない。

# New path
def assign_driver(waiting: Waiting, driver_id: UUID, now: datetime) -> EnRoute: ...

# Legacy wrapper during migration
class TaxiRequestService:
    def assign_driver(self, request_id: UUID, driver_id: UUID) -> None:
        row = self.repo.get(request_id)
        waiting = row_to_waiting(row)
        en_route = assign_driver(waiting, driver_id, datetime.now(UTC))
        self.repo.save(en_route.model_dump(mode="python"))

呼び出し側がユースケースへ移ったらラッパーを削除する。

フェーズ 3: ユースケースを抽出

サービスメソッドをポートを受け取る非同期関数にする。完全実装: 状態遷移。

async def assign_driver_use_case(
    resolver: RequestResolver,
    store: RequestStore,
    request_id: UUID,
    driver_id: UUID,
    now: datetime,
) -> Result[EnRoute, AssignDriverError]:
    ...

コントローラーはユースケースを呼ぶ。レガシーサービスは削除まで委譲する。

フェーズ 4: リポジトリプロトコル

SQLAlchemy/Django ORM クエリをアダプターモジュールへ移す。ユースケースが扱うのはドメイン状態と明示的なエラーだけにすべきだ。

mapped_column エンティティ、行 DTO、domain_from_row_dto マッパー、Django select_for_update 書き込みパターンは ORM アダプターを読む。

フェーズ 5: イベントとアウトボックス

監査、統合、非同期反応が必要なワークフローにだけドメインイベントを追加する。1 イベント型と 1 コンシューマーから始める。

共存ルール

新旧スタイルが共存している間:

新コードは変更中のワークフローに Kamae Python に従う。
未変更レガシーは即時マイグレーション不要。
混ぜない: 同一集約について、blob 状態モデルと判別共用体を競合する真実の源として併用しない。
文書化: 一時ラッパーに短いコメントを付け、可能なら同一エピックで削除する。

段階的型安全性

mypy カバレッジを徐々に拡大する:

[tool.mypy]
files = [
    "src/taxi_request",
    "tests/taxi_request",
]

移行したディレクトリを追加する。グローバル --strict の前に移行パッケージで check_kamae_policy.py を使う。

してはいけないこと

次は移行が長期化しやすい典型パターンである。症状に心当たりがあれば、一度立ち止まってスライスを小さくする。

1 垂直スライスを出す前にすべてのモデルを書き換える（「半年フリーズ」型の全面書き換え）
マイグレーションの見た目のためだけに DI フレームワークを導入する
境界が型付けされる前にすべてのレガシーメソッドに Result を強制する
ワークフローがまだ CRUD だけ必要なのに、完全アウトボックス/イベントインフラでリリースをブロックする

1 ワークフロー移行の成功基準

「全体の移行が終わった」ではなく、触った 1 ワークフローが観測可能に Kamae 準拠になったことを基準にする。次が揃えば、そのスライスはレビューと本番投入の候補になる。

外部入力が境界で Pydantic によりパースされる
アグリゲート状態が frozen 判別共用体として表現される
ビジネス遷移がテスト付き純粋関数である
ユースケースが明示的エラーを返すか、インフラ失敗を明確にマップする
リポジトリアダプターが必要なら状態を原子性保存する
可観測性が PII と観測経路の保護に従う

最初に移行するワークフローの選択は永続化、集約、イベントとアプリケーション配線を読む。

レビュー観点

移行はオブザーバビリティと PII 姿勢を保つか — High

移行経路が生ペイロードの旧ログを維持し、マスクを落とし、新設計で必要なトランザクション/アウトボックス保証を迂回する箇所を指摘する。

PII と観測経路の保護と永続化、集約、イベントと照合する。

レガシーコードは明確に隔離されているか — Medium

文書化された境界なしに、新しい Kamae スタイルモジュールが旧層の型付きでない dict、可変グローバル、ORM エンティティに依存する箇所を指摘する。

差分は全面書き換えの前に境界を改善しているか — Medium

触れたワークフローで DTO パース、状態型付け、エラーマッピングを先に締めずにレガシーサービスクラスを動かす大規模書き換えを指摘する。

互換シムは薄く一時的か — Low

利便のためドメインロジックを恒久的に二重化したり無効状態を保持したりする広いアダプター層を指摘する。