新旧データの「同じもの」を、Identity型とDomainで判定する

対応判定はRepositoryではなくDomainに置く

システムを段階的に更新していると、新しい識別子を持つデータと、それを持たない過去データが同時に存在する期間があります。このとき必要なのは、単に両方を取得することではありません。識別方法の異なる二つのレコードを、どの条件なら「同じもの」とみなしてよいかを決めることです。

この問題では、Repository は比較候補を取得するところまでとし、新旧データの対応判定は Domain に置くようにしました。

図を読み込み中...

Domain では、新旧それぞれの識別方法を条件分岐のまま持たず、Identity という値に変換します。Identity は通常使う primary_key と、過去データとの互換性を保つ fallback_key を持ちます。

この形にすると、旧データから新データを探す場合も、新データから旧データを探す場合も、同じマッチング規則を使えます。形式ごとの分岐を増やすのではなく、同一性の表現と対応判定を共通化できることが、この設計の利点です。

なぜRepositoryだけでは解決しきれないのか

Repository は、条件に合うレコードを DB から取得したり、永続化形式をアプリケーションの型へ変換したりする場所です。完全一致するキーがあるなら、Repository や SQL の JOIN だけでも扱えます。

しかし、新旧データの橋渡しには次の判断が含まれていました。

• 新しい識別子があれば、そちらの一致を優先する
• 新旧をまたぐ場合だけ、互換用のキーへfallbackする
• fallbackの候補が複数あるなら、推測で一件に決めない
• 比較の向きが逆でも、同じ安全性を保つ

これらは「どの行を取得するか」ではなく、二つのデータを業務上同じものと扱ってよいかという判断です。SQL に書くこと自体はできますが、書けることと、その責務を Repository に置くことは別です。

今回の役割分担は、次のようにしました。

Repository にマッピングまで持たせると、取得方法と対応判定が一体になります。将来 DB の取得方法だけを変えたい場合にも、旧データとの互換ルールを読み解かなければなりません。反対に Domain へ分けると、候補をメモリ、DB、別APIのどこから取得しても、対応判定そのものは変えずに使えます。

Identity型にprimary keyとfallback keyを持たせる

新旧データを一つの比較処理へ渡すために、まず各レコードから不変な Identity を作ります。

from dataclasses import dataclass

IdentityKey = tuple[str, int, str]
FallbackKey = tuple[str, int, str]

@dataclass(frozen=True)
class Identity:
    primary_key: IdentityKey
    fallback_key: FallbackKey

primary_key は、そのデータ形式で最も強く識別できるキーです。新データでは追加された識別子を使い、旧データでは従来の属性を使います。

一方の fallback_key は、新旧の両方から作れる互換用のキーです。新データにもあえてこのキーを持たせることで、旧データと比較するときだけ同じ土俵へ降りられます。

記事用に商品データへ置き換えると、生成処理は次のような形です。

def build_identity(
    item: ItemRecord,
    checked_at: datetime,
) -> Identity:
    normalized_label = normalize_label(item.label)
    fallback_key = ("sku_label", item.sku_id, normalized_label)

    if should_use_variant_identity(checked_at):
        variant_id = normalize_variant_id(item.variant_id)

        if variant_id is not None:
            return Identity(
                primary_key=("variant", item.sku_id, variant_id),
                fallback_key=fallback_key,
            )

    return Identity(
        primary_key=("legacy", item.sku_id, normalized_label),
        fallback_key=fallback_key,
    )

ここで良かったのは、if old ... else new ... という分岐を、比較処理のあちこちへ持ち込まなかったことです。形式差は Identity を組み立てる時点で吸収し、その後の処理は二つのキーだけを見ます。

これは、振る舞いを何でもクラスへ詰め込む設計ではありません。Identity は比較に必要な意味を持つ小さなデータで、生成と解決は副作用のない関数として扱います。関数型のイディオムを大げさに導入せず、データを正規化してから同じ関数へ流す考え方だけを使っています。

同じ規則で旧から新へ、新から旧へ対応付ける

Identity を作ると、比較の向きごとに別のアルゴリズムを書く必要がなくなります。

図を読み込み中...

比較対象は、primary key と、一意な fallback key で引ける lookup にしておきます。fallback key が複数件に対応する場合、そのキーは unique_by_fallback へ入れません。

@dataclass(frozen=True)
class IdentityLookup:
    by_primary: Mapping[IdentityKey, ItemRecord]
    unique_by_fallback: Mapping[FallbackKey, ItemRecord]

解決順序は、完全一致を優先し、それがなければfallbackを試すだけです。ただし、橋渡しの前に比較元と比較先の両方で一意性を確認します。

def resolve_matching_item(
    source: ItemWithIdentity,
    source_lookup: IdentityLookup,
    target_lookup: IdentityLookup,
) -> ItemRecord | None:
    exact_match = target_lookup.by_primary.get(source.identity.primary_key)

    if exact_match is not None:
        return exact_match

    fallback_key = source.identity.fallback_key

    if fallback_key not in source_lookup.unique_by_fallback:
        return None

    return target_lookup.unique_by_fallback.get(fallback_key)

この関数へ旧データを source として渡せば旧から新を探せます。引数を入れ替えて新データを source にすれば、新から旧も同じ規則で探せます。

図を読み込み中...

この「両側で一意なときだけ結ぶ」が重要です。たとえば、一つの旧データに対応しそうな新データが二件あれば、どちらかを選ぶ根拠はありません。逆向きでも同じです。シームレスに扱うとは、曖昧さを隠して自動的に結ぶことではなく、向きによらず同じ安全基準で結果を出せることだと考えました。

ロールアウト日はIdentity生成の補助条件にする

ロールアウト日の判定も必要ですが、今回の設計の中心ではありません。日付は「どちらの primary_key を作るか」を決める補助条件です。

def should_use_variant_identity(checked_at: datetime) -> bool:
    checked_on = checked_at.astimezone(APP_TIMEZONE).date()
    return checked_on >= ROLLOUT_DATE

現在時刻ではなくデータ自身に記録された日時を使い、境界日とタイムゾーンの扱いを一箇所へ閉じ込めました。比較処理は日付を知らず、完成した Identity だけを受け取ります。

Domainに置くことで何が良くなったか

この分割によって、単にファイルの置き場所が変わったわけではありません。変更理由が分かれました。

図を読み込み中...

• 取得条件やDB構成が変わった場合は、Repositoryを変更する
• APIの入出力や処理順が変わった場合は、Serviceを変更する
• 同一性、fallback、一意性の定義が変わった場合は、Domainを変更する

特に、Domain が SQLAlchemy、Pydantic、FastAPI を知らないことが効きました。マッピング規則を小さな入力だけでテストでき、レビューでも「この二件を同じものと扱ってよいか」という本来の論点へ集中できます。

Repository へ複雑な JOIN を追加して結果だけ返す設計では、なぜ候補が除外されたのか、どの曖昧さを避けたのかが見えにくくなります。Domain の Identity と解決関数として表すことで、その判断をコード上の仕様として残せました。

テストはマッピング仕様として並べる

テストも、ロールアウト日の前後より、Identity とマッピングの組み合わせを中心にしました。

TestBuildIdentity
  ├ test_builds_new_primary_key_with_legacy_fallback
  ├ test_builds_legacy_primary_key_before_rollout
  └ test_treats_empty_variant_id_as_missing

TestResolveMatchingItem
  ├ test_primary_key_match_takes_priority
  ├ test_maps_old_to_new_when_both_sides_are_unique
  ├ test_maps_new_to_old_when_both_sides_are_unique
  ├ test_rejects_ambiguous_source_fallback
  └ test_rejects_ambiguous_target_fallback

この並びにすると、テスト名自体が新旧マッピングの仕様になります。DB や HTTP を起動せずに確認できるため、fallbackの条件を変更するときも影響範囲を追いやすくなりました。

この設計で気をつけたいこと

この設計なら何でも Domain へ移せばよい、という話ではありません。特に次の三点は明示しておく必要があります。

1. fallback keyにはドメイン上の意味が必要です。 たまたまDBにあるカラムを連結しただけでは、Repositoryの都合をDomainへ移しただけになります。「なぜこの組み合わせなら過去データと同じ可能性があるのか」を説明できる属性を使います。
2. 曖昧さを呼び出し側が区別したいなら、Noneだけで返しません。 未一致と候補複数で後続処理が変わる場合は、Matched、NotFound、Ambiguous のような結果型にします。
3. fallbackには終了条件が必要です。 過去データの保持期間が終わった後も互換ロジックを残すと、Identityの意味が複雑なまま固定されます。いつ削除できるかも移行設計に含めます。

重要なのは「Domain層を作ること」そのものではありません。Repository の取得責務と、データを同一とみなす意味判断を分け、その判断を小さな型と純粋な関数で読める形にすることです。

まとめ

新旧データのマッピングでは、Repository が両方のデータを取得できても、それだけでは対応関係を決められません。完全一致を優先するのか、いつfallbackしてよいのか、候補が複数ならどうするのかは、ドメインの判断です。

今回の設計では、次の形に整理しました。

• Repositoryは比較候補の取得に留める
• Domainで各データをIdentity(primary_key, fallback_key)へ変換する
• primary keyの完全一致を優先する
• fallbackは比較元と比較先の両方で一意な場合だけ使う
• 同じ解決関数で旧から新、新から旧の両方を扱う
• ロールアウト日はIdentityを作るための補助条件に留める

この形なら、新旧の形式差をserviceやrepositoryの分岐として広げず、「同じものとみなせるか」という判断を一箇所で説明し、テストできます。新旧データを扱うときは、取得方法を考えるだけでなく、同一性をどの層が判断するのかを先に切り分けることが重要です。