A コーディング — 単一責任原則・継承より委譲・スマートUI/神クラス対処（キャンペーン対象判定 PromotionEligibilityService Bad→Good Ch6-7）

概要

問題

EC企業 MOps 販促システムチームでは、注文が各種キャンペーン（初回購入限定・優良顧客ランク限定・クーポン併用可）の対象かどうかを判定する PromotionEligibilityService が本番稼働している。判定に加えて、Slackへの結果通知・BigQueryへの監査ログ書き込みまでを同じクラス階層が担っており、次のインシデントが起きた。

「良いコード・悪いコードで学ぶ設計入門」第6-7章（関心の分離：単一責任原則、継承より委譲、スマートUI・神クラス対処）の考え方を使って、PromotionEligibilityServiceを再設計してください。

制約・前提条件

• Python 3.12+
• 判定対象のキャンペーンは「初回購入限定」「優良顧客ランク限定」「クーポン併用可」の3種類。将来的にキャンペーン種別が追加される前提で設計すること
• 判定ロジックは「顧客・注文の情報」だけから純粋に結果を返せるようにし、Slack通知・BigQuery監査ログ書き込みという副作用を判定ロジックから分離すること
• Slack通知・BigQuery監査ログ書き込みは外部サービスへの実アクセスを伴うため、ユニットテストではモック（フェイク実装）に差し替えられる設計にすること
• 判定結果のキャッシュは「顧客ID」だけでなく「キャンペーン種別」も区別してキー化し、インシデントを再発させないこと
• 新しいキャンペーン種別を追加する際に、通知・監査ログを担うクラスへの変更が発生しない設計にすること

悪いコード (Before)

📋 コピーclass PromotionEligibilityService:
    _cache = {}  # 顧客IDのみでキー化、キャンペーン種別を無視

    def __init__(self, customer_id, order_total):
        self.customer_id = customer_id
        self.order_total = order_total
        self.slack_client = SlackWebhookClient(url=SLACK_WEBHOOK_URL)
        self.bq_client = bigquery.Client()

    def check(self):
        raise NotImplementedError

    def send_slack_notification(self, message):
        self.slack_client.post(message)

    def write_audit_log(self, result):
        table = self.bq_client.dataset("mops").table("promo_audit_log")
        self.bq_client.insert_rows(table, [{
            "customer_id": self.customer_id,
            "result": result,
        }])

    def run(self):
        if self.customer_id in self._cache:
            return self._cache[self.customer_id]
        result = self.check()
        self._cache[self.customer_id] = result
        self.write_audit_log(result)
        self.send_slack_notification(f"{self.customer_id}: {result}")
        return result


class FirstPurchasePromotion(PromotionEligibilityService):
    def check(self):
        orders = fetch_past_orders(self.customer_id)  # 重複実装①
        return len(orders) == 0 and self.order_total >= 3000


class LoyaltyTierPromotion(PromotionEligibilityService):
    def check(self):
        orders = fetch_past_orders(self.customer_id)  # 重複実装②
        tier = fetch_member_tier(self.customer_id)
        return tier in ("gold", "platinum") and len(orders) >= 5


class CouponStackPromotion(PromotionEligibilityService):
    def check(self):
        orders = fetch_past_orders(self.customer_id)  # 重複実装③（未使用）
        return self.order_total >= 5000

# --- 呼び出し側 ---
promo = FirstPurchasePromotion(customer_id="C001", order_total=4000)
promo.run()  # => True

promo2 = LoyaltyTierPromotion(customer_id="C001", order_total=8000)
promo2.run()  # => True のはずが、_cache["C001"]が再利用されてバグる

ヒント（段階的開示）

from __future__ import annotations
from dataclasses import dataclass
from typing import Protocol

@dataclass(frozen=True, slots=True)
class OrderContext:
    customer_id: str
    order_total: int
    past_order_count: int
    member_tier: str | None

class EligibilityRule(Protocol):
    name: str
    def is_eligible(self, context: OrderContext) -> bool: ...

@dataclass(frozen=True, slots=True)
class FirstPurchaseRule:
    name: str = "first_purchase"
    min_order_total: int = 3000
    def is_eligible(self, context: OrderContext) -> bool:
        return ...

class NotificationPort(Protocol):
    def notify(self, message: str) -> None: ...

class AuditLogPort(Protocol):
    def record(self, customer_id: str, rule_name: str, result: bool) -> None: ...

class PromotionEligibilityChecker:
    def __init__(self, notifier: NotificationPort, audit_log: AuditLogPort) -> None:
        self._notifier = notifier
        self._audit_log = audit_log
        self._cache: dict[tuple[str, str], bool] = {}

    def run(self, rule: EligibilityRule, context: OrderContext) -> bool:
        cache_key = (context.customer_id, rule.name)
        ...

問題点分析（7点）

構造の比較（SVG）— 継承階層 vs 委譲による組み立て

模範解答

📋 コピー"""キャンペーン対象判定モジュール。

良いコード・悪いコードで学ぶ設計入門（改訂新版）
  Ch6-7: 関心の分離 — 単一責任原則、継承より委譲、スマートUI・神クラス対処

判定ロジック(EligibilityRule)・通知(NotificationPort)・監査ログ(AuditLogPort)の
3つの責務をそれぞれ別のオブジェクトに分離し、PromotionEligibilityCheckerは
それらを「組み合わせる」役割だけを持つ。継承ではなく委譲(コンストラクタ注入)で
振る舞いを組み立てることで、キャンペーン種別が増えてもCheckerは変更不要になる。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Protocol

# キャッシュ未登録を判定するための番兵オブジェクト（Noneが正当な値と衝突しないようにする）
_CACHE_UNSET = object()


@dataclass(frozen=True, slots=True)
class OrderContext:
    """キャンペーン対象判定に必要な入力をまとめた値オブジェクト。

    Attributes:
        customer_id: 判定対象の顧客ID。
        order_total: 今回の注文金額（円）。
        past_order_count: 過去の注文件数。
        member_tier: 会員ランク（"gold" / "platinum" 等）。未加入はNone。
    """
    customer_id: str
    order_total: int
    past_order_count: int
    member_tier: str | None = None


class EligibilityRule(Protocol):
    """キャンペーン対象判定ルールのインターフェース。

    判定のみを行い、通知やログ書き込みといった副作用を一切持たない
    （純粋関数的なオブジェクトとして扱う）。
    """
    name: str

    def is_eligible(self, context: OrderContext) -> bool:
        """contextがこのキャンペーンの対象条件を満たすかを返す。"""
        ...


@dataclass(frozen=True, slots=True)
class FirstPurchaseRule:
    """初回購入限定キャンペーンの対象判定ルール。"""
    name: str = "first_purchase"
    min_order_total: int = 3000  # 対象となる最低注文金額（円）

    def is_eligible(self, context: OrderContext) -> bool:
        return (
            context.past_order_count == 0
            and context.order_total >= self.min_order_total
        )


@dataclass(frozen=True, slots=True)
class LoyaltyTierRule:
    """優良顧客ランク限定キャンペーンの対象判定ルール。"""
    name: str = "loyalty_tier"
    eligible_tiers: tuple[str, ...] = field(default=("gold", "platinum"))
    min_past_orders: int = 5  # 対象となる最低の過去注文件数

    def is_eligible(self, context: OrderContext) -> bool:
        return (
            context.member_tier in self.eligible_tiers
            and context.past_order_count >= self.min_past_orders
        )


@dataclass(frozen=True, slots=True)
class CouponStackRule:
    """クーポン併用可キャンペーンの対象判定ルール。"""
    name: str = "coupon_stack"
    min_order_total: int = 5000  # 対象となる最低注文金額（円）

    def is_eligible(self, context: OrderContext) -> bool:
        return context.order_total >= self.min_order_total


class NotificationPort(Protocol):
    """判定結果の通知先を抽象化するインターフェース。"""
    def notify(self, message: str) -> None: ...


class AuditLogPort(Protocol):
    """判定結果の監査ログ書き込み先を抽象化するインターフェース。"""
    def record(self, customer_id: str, rule_name: str, result: bool) -> None: ...


class SlackNotificationAdapter:
    """NotificationPortのSlack実装（本番用）。"""

    def __init__(self, client: SlackWebhookClient) -> None:
        self._client = client

    def notify(self, message: str) -> None:
        self._client.post(message)


class BigQueryAuditLogAdapter:
    """AuditLogPortのBigQuery実装（本番用）。"""

    def __init__(self, bq_client: bigquery.Client, table_ref: str) -> None:
        self._bq_client = bq_client
        self._table_ref = table_ref

    def record(self, customer_id: str, rule_name: str, result: bool) -> None:
        self._bq_client.insert_rows(
            self._table_ref,
            [{
                "customer_id": customer_id,
                "rule_name": rule_name,
                "result": result,
                "checked_at": datetime.now(timezone.utc).isoformat(),
            }],
        )


class PromotionEligibilityChecker:
    """キャンペーン対象判定を組み合わせて実行する、単一責任のオーケストレータ。

    判定ロジック(EligibilityRule)・通知(NotificationPort)・監査ログ(AuditLogPort)
    のいずれも自身では実装せず、コンストラクタで注入されたコラボレータへ委譲する。
    このクラスの変更理由は「判定結果の組み合わせ方（キャッシュ・記録の手順）」のみに
    限定され、新しいキャンペーン種別の追加や通知先の変更ではこのクラスは変更されない。

    Attributes:
        _notifier: 判定結果を通知するためのポート。
        _audit_log: 判定結果を記録するためのポート。
        _cache: (顧客ID, ルール名)をキーにした判定結果キャッシュ。
            キャンペーン種別をまたいだ誤共有を防ぐため、ルール名を必ずキーに含める。
    """

    def __init__(self, notifier: NotificationPort, audit_log: AuditLogPort) -> None:
        self._notifier = notifier
        self._audit_log = audit_log
        self._cache: dict[tuple[str, str], bool] = {}

    def run(self, rule: EligibilityRule, context: OrderContext) -> bool:
        """指定ルールで対象判定を行い、通知・監査ログを1回だけ記録する。

        Args:
            rule: 適用するキャンペーン判定ルール。
            context: 判定に必要な顧客・注文情報。

        Returns:
            キャンペーン対象であればTrue。
        """
        cache_key = (context.customer_id, rule.name)
        cached = self._cache.get(cache_key, _CACHE_UNSET)
        if cached is not _CACHE_UNSET:
            return cached  # type: ignore[return-value]

        result = rule.is_eligible(context)
        self._cache[cache_key] = result
        self._audit_log.record(context.customer_id, rule.name, result)
        self._notifier.notify(f"{context.customer_id} / {rule.name}: {result}")
        return result

📋 コピー# --- 本番相当の組み立て ---
checker = PromotionEligibilityChecker(
    notifier=SlackNotificationAdapter(slack_client),
    audit_log=BigQueryAuditLogAdapter(bq_client, "mops.promo_audit_log"),
)

ctx1 = OrderContext(customer_id="C001", order_total=4000, past_order_count=0)
print(checker.run(FirstPurchaseRule(), ctx1))
# => True （Slack通知・BigQuery監査ログ書き込みがそれぞれ1回だけ実行される）

# 同一顧客に対して別キャンペーン種別を判定
ctx2 = OrderContext(customer_id="C001", order_total=8000, past_order_count=6, member_tier="gold")
print(checker.run(LoyaltyTierRule(), ctx2))
# => True
# 旧実装ではキャッシュキーが顧客IDのみだったため、直前のFirstPurchaseRuleの
# 判定結果(True)がそのまま誤って返っていた（インシデントの再現）。
# 新実装は(customer_id, rule.name)でキー化しているため、別ルールとして正しく再判定される。

# --- テスト時: 実際のSlack/BigQueryにアクセスせずに検証 ---
class FakeNotifier:
    def __init__(self) -> None:
        self.messages: list[str] = []
    def notify(self, message: str) -> None:
        self.messages.append(message)

class FakeAuditLog:
    def __init__(self) -> None:
        self.records: list[tuple[str, str, bool]] = []
    def record(self, customer_id: str, rule_name: str, result: bool) -> None:
        self.records.append((customer_id, rule_name, result))

test_checker = PromotionEligibilityChecker(notifier=FakeNotifier(), audit_log=FakeAuditLog())
ctx3 = OrderContext(customer_id="C999", order_total=6000, past_order_count=2)
assert test_checker.run(CouponStackRule(), ctx3) is True

ポイント解説

実務への応用

MOps 販促システムチームでは、キャンペーン種別が今後も増える見込みであり、都度サブクラスを追加する設計のままでは「新しいキャンペーンを追加するたびに、既存の通知・監査ログ処理まで意図せず影響を受けないか」をレビューでいちいち確認する必要が生じる。EligibilityRuleを委譲で組み立てる設計にしておけば、新しいキャンペーン種別の追加は「Ruleクラスを1つ実装してChecker.run()に渡す」だけで完結し、既存コードへの変更差分をレビュー時に確認しやすくなる。

今日のまとめ

次のステップ

• 発展問題: EligibilityRuleを複数組み合わせて「初回購入 かつ クーポン併用可」のようにAND/OR条件でキャンペーンを合成できるようにするCompositeRuleを、継承ではなく委譲（Composite/Decoratorパターン）で設計する
• 参考: 「良いコード・悪いコードで学ぶ設計入門」第6章（関心の分離）、第7章（継承より委譲・スマートUI・神クラス）、第8章（条件分岐・ストラテジパターン、EligibilityRuleと関連が深い）

自己評価（あとで記入）