A コーディング — 継承より委譲・神クラス対処（在庫連携ステータス通知サービス Bad→Good Ch7）

概要

問題

EC企業 MOps 販促システムチームでは、夜間バッチ（Argo Workflows）で実行する在庫連携処理の結果（成功件数・失敗件数）を、Slack・Email・LINE Notifyの3チャネルへ通知するInventorySyncNotifierを運用している。

このクラスには2つの構造的な問題がある。

1. 神クラス（God Class）: InventorySyncNotifierが「送信対象チャネルの判定（if/elif）」「メッセージ整形」「送信」「リトライ」「BigQueryへのログ記録」を単独で全て抱えている
2. 継承の誤用: LineNotifierはSlackNotifierとは通知先も送信形式もまったく異なる（is-a関係が成立しない）にもかかわらず、リトライロジックだけを再利用する目的でSlackNotifierを継承している。そのためLineNotifierは「Slack用のwebhook URLフィールド」を意味もなく引き継ぎ、notify()メソッドは親クラスの実装を一切使わず全部上書きしている

「良いコード・悪いコードで学ぶ設計入門」第7章（継承より委譲、スマートUIおよび神クラスの対処法）を使って、この設計をリファクタリングしてください。

制約・前提条件

• Python 3.12+ を前提とする（Protocol, StrEnum, dataclass(slots=True), asyncio.TaskGroup などモダン記法を使ってよい）
• LineNotifierがSlackNotifierを継承する設計を廃止し、継承を使わないこと。チャネル間で共通する処理（リトライ）は、独立したオブジェクトへの委譲で共有すること
• 新しい通知チャネル（例: Microsoft Teams）を追加する際に、既存クラスのif/elifチェーンを1行も変更せずに追加できる設計にすること（Open/Closed Principle）
• Slack・Email・LINEへの送信は並行実行し、1チャネルの遅延が他チャネルの通知をブロックしないこと
• SyncResult（バッチ結果）は@dataclass(frozen=True, slots=True)で不変オブジェクトとして表現すること
• Google スタイル docstring・インラインコメント・型ヒントを含めること

悪いコード (Before)

📋 コピーimport time
import requests


class SlackNotifier:
    def __init__(self, webhook_url):
        self.webhook_url = webhook_url

    def _send_with_retry(self, payload, max_retries=3):
        for attempt in range(max_retries):
            try:
                resp = requests.post(self.webhook_url, json=payload, timeout=5)
                if resp.status_code == 200:
                    return
            except Exception:
                pass
            time.sleep(2 ** attempt)
        raise RuntimeError("Slack通知に失敗しました")

    def notify(self, sync_result):
        payload = {"text": f"在庫連携: 成功{sync_result['success_count']}件"}
        self._send_with_retry(payload)


class LineNotifier(SlackNotifier):  # 問題1: リトライ再利用だけが目的の継承
    def __init__(self, line_token):
        self.webhook_url = "https://notify-api.line.me/api/notify"
        self.line_token = line_token

    def notify(self, sync_result):
        # 問題2: LINEはSlackとpayload形式が違うため親のnotify()を使えず全部上書き
        payload = f"message=在庫連携完了 成功{sync_result['success_count']}件"
        headers = {"Authorization": f"Bearer {self.line_token}"}
        for attempt in range(3):  # 問題3: リトライロジックがSlackとほぼ同じ内容でコピペ
            try:
                resp = requests.post(
                    self.webhook_url, data=payload, headers=headers, timeout=5
                )
                if resp.status_code == 200:
                    return
            except Exception:
                pass
            time.sleep(2 ** attempt)
        raise RuntimeError("LINE通知に失敗しました")


class InventorySyncNotifier:  # 問題4: 神クラス。取得/整形/送信/リトライ/ログを全部保持
    def __init__(self, slack_webhook, line_token, email_client, bq_client):
        self.slack = SlackNotifier(slack_webhook)
        self.line = LineNotifier(line_token)
        self.email_client = email_client
        self.bq_client = bq_client

    def notify_all(self, sync_result, channels):
        for ch in channels:  # 問題5: 新チャネル追加のたびif/elifが肥大化
            if ch == "slack":
                self.slack.notify(sync_result)
            elif ch == "line":
                self.line.notify(sync_result)
            elif ch == "email":
                for attempt in range(1):  # 問題6: Emailだけリトライ回数がコピペで不統一
                    try:
                        self.email_client.send(
                            to="mops-team@example.com",
                            subject="在庫連携結果",
                            body=f"成功{sync_result['success_count']}件",
                        )
                        break
                    except Exception:
                        time.sleep(1)
            self.bq_client.insert("sync_notification_log", {"channel": ch})
        # 問題7: 同期forループのため、Slackのレート制限待ちがEmail/LINE送信を
        # 丸ごとブロックし、障害検知の通知が遅延する

ヒント（段階的開示）

class NotificationChannel(Protocol):
    async def send(self, result: SyncResult) -> None: ...

class RetryPolicy:
    def __init__(self, max_retries: int = 3) -> None:
        self._max_retries = max_retries
    async def run(self, send_once) -> None:
        ...  # 指数バックオフでsend_once()をリトライ

class SlackChannel:
    def __init__(self, webhook_url: str, retry_policy: RetryPolicy) -> None:
        self._webhook_url = webhook_url
        self._retry_policy = retry_policy  # 継承ではなく委譲でリトライを共有

class InventorySyncDispatcher:
    def __init__(self, channels: dict[ChannelType, NotificationChannel], bq_client) -> None:
        self._channels = channels  # if/elifではなく辞書で拡張可能に

問題点分析（7点）

構造の比較（SVG）— 継承による責務の混線 vs Protocol+委譲での分離

模範解答

📋 コピー"""inventory_sync_notification.py — 継承より委譲・スマートUI/神クラス対処（Ch7）

在庫連携バッチ(Argo Workflows)の結果をSlack/Email/LINEへ通知する処理を、
「神クラスが全チャネルの送信ロジックを持つ」「LineNotifierがSlackNotifierを
リトライ再利用のためだけに継承する」という2つの問題から、Protocol(構造的型付け)
による委譲ベースの設計へ置き換える。
"""
from __future__ import annotations

import asyncio
from dataclasses import dataclass
from enum import StrEnum
from typing import Final, Protocol

_DEFAULT_MAX_RETRIES: Final[int] = 3
_BACKOFF_BASE_SECONDS: Final[float] = 2.0
_EMAIL_MAX_RETRIES: Final[int] = 1  # メール送信は冪等性が保証されないため再送を最小限にする


class ChannelType(StrEnum):
    """通知チャネルの種別。"""

    SLACK = "slack"
    EMAIL = "email"
    LINE = "line"


@dataclass(frozen=True, slots=True)
class SyncResult:
    """在庫連携バッチの実行結果(不変)。"""

    success_count: int
    fail_count: int
    batch_id: str


class NotificationChannel(Protocol):
    """通知チャネルが満たすべき契約。継承ではなく構造的型付けで多態性を得る(修正1・4)。"""

    async def send(self, result: SyncResult) -> None:
        """整形済みメッセージを送信する。失敗時は例外をraiseする。"""
        ...


class RetryPolicy:
    """リトライ回数・バックオフを1箇所に集約し、各チャネルへ委譲で提供する(修正1・3・6)。

    継承で共有すると親クラスの変更が無関係な子クラスに波及するため、
    独立したオブジェクトとして各チャネルのコンストラクタで受け取らせる。
    """

    def __init__(self, max_retries: int = _DEFAULT_MAX_RETRIES) -> None:
        self._max_retries = max_retries

    async def run(self, send_once) -> None:
        """send_once(引数なしの非同期callable)を指数バックオフ付きでリトライ実行する。

        Args:
            send_once: 1回分の送信処理を行う非同期callable。

        Raises:
            RuntimeError: 最大リトライ回数を超えても成功しなかった場合。
        """
        last_error: Exception | None = None
        for attempt in range(self._max_retries):
            try:
                await send_once()
                return
            except Exception as exc:  # noqa: BLE001 - 送信先固有の例外は一旦集約しログへ残す
                last_error = exc
                await asyncio.sleep(_BACKOFF_BASE_SECONDS**attempt)
        raise RuntimeError(f"送信に{self._max_retries}回失敗しました") from last_error


class SlackChannel:
    """Slack Webhookへの送信を担当する(継承なし、独立したクラス)(修正1)。"""

    def __init__(self, webhook_url: str, retry_policy: RetryPolicy) -> None:
        self._webhook_url = webhook_url
        self._retry_policy = retry_policy  # 修正2: リトライは委譲で共有

    async def send(self, result: SyncResult) -> None:
        payload = {"text": f"在庫連携: 成功{result.success_count}件, 失敗{result.fail_count}件"}

        async def _once() -> None:
            await self._post(json=payload)

        await self._retry_policy.run(_once)

    async def _post(self, *, json: dict) -> None:
        ...  # 実際のhttpx.AsyncClient呼び出し(略)


class LineChannel:
    """LINE Notifyへの送信を担当する。SlackChannelとは無関係な独立クラス(修正1)。"""

    def __init__(self, line_token: str, retry_policy: RetryPolicy) -> None:
        self._line_token = line_token
        self._retry_policy = retry_policy  # 修正2: SlackChannelを継承せず同じ仕組みを委譲で共有

    async def send(self, result: SyncResult) -> None:
        payload = f"message=在庫連携完了 成功{result.success_count}件"

        async def _once() -> None:
            await self._post(payload)

        await self._retry_policy.run(_once)

    async def _post(self, payload: str) -> None:
        ...  # 実際のLINE Notify API呼び出し(略)


class EmailChannel:
    """メール通知を担当する。API冪等性が低いためリトライ回数を絞る(修正6)。"""

    def __init__(self, email_client, retry_policy: RetryPolicy) -> None:
        self._email_client = email_client
        self._retry_policy = retry_policy  # EmailChannel専用にmax_retries=1のRetryPolicyを渡せる

    async def send(self, result: SyncResult) -> None:
        async def _once() -> None:
            await self._email_client.send(
                to="mops-team@example.com",
                subject="在庫連携結果",
                body=f"成功{result.success_count}件 失敗{result.fail_count}件",
            )

        await self._retry_policy.run(_once)


class InventorySyncDispatcher:
    """複数チャネルへ結果を配信する。チャネル固有の実装は一切持たず委譲するだけ(修正4・5)。"""

    def __init__(self, channels: dict[ChannelType, NotificationChannel], bq_client) -> None:
        self._channels = channels  # 修正5: if/elifチェーンではなく辞書で拡張可能にする
        self._bq_client = bq_client

    async def notify_all(self, result: SyncResult, targets: list[ChannelType]) -> None:
        """指定チャネルへ並行して送信する(修正7: TaskGroupで並行化)。

        Args:
            result: 通知するバッチ結果。
            targets: 送信対象チャネル一覧。

        Raises:
            ExceptionGroup: いずれかのチャネル送信が失敗した場合(TaskGroupが集約)。
        """
        async with asyncio.TaskGroup() as tg:  # 修正7: 順次forループではなく並行実行
            for channel_type in targets:
                channel = self._channels[channel_type]
                tg.create_task(self._send_and_log(channel_type, channel, result))

    async def _send_and_log(
        self, channel_type: ChannelType, channel: NotificationChannel, result: SyncResult
    ) -> None:
        await channel.send(result)
        await self._bq_client.insert(
            "sync_notification_log",
            {"channel": channel_type.value, "batch_id": result.batch_id},
        )

📋 コピーretry_policy = RetryPolicy(max_retries=_DEFAULT_MAX_RETRIES)
email_retry_policy = RetryPolicy(max_retries=_EMAIL_MAX_RETRIES)

dispatcher = InventorySyncDispatcher(
    channels={
        ChannelType.SLACK: SlackChannel("https://hooks.slack.com/xxx", retry_policy),
        ChannelType.LINE: LineChannel("line-token-xxx", retry_policy),
        ChannelType.EMAIL: EmailChannel(email_client, email_retry_policy),
    },
    bq_client=bq_client,
)

result = SyncResult(success_count=980, fail_count=20, batch_id="sync-2026-09-07-01")
await dispatcher.notify_all(
    result, targets=[ChannelType.SLACK, ChannelType.LINE, ChannelType.EMAIL]
)
# => Slack/LINE/Emailへ「同時に」送信が開始される。
#    Slackがレート制限でリトライ待ちになっても、LINE/Emailの送信はブロックされない。
# => BigQueryのsync_notification_logに3チャネル分のログが記録される。
#
# 新チャネル(Teams)を追加する場合:
#   1. TeamsChannel(NotificationChannel)を1つ実装する
#   2. channels辞書に ChannelType.TEAMS: TeamsChannel(...) を1行追加する
#   → InventorySyncDispatcherのコードは1行も変更しない

ポイント解説

実務への応用

• MOps販促システムチームでは、キャンペーン結果通知・在庫連携結果通知・A/Bテスト集計完了通知など「複数の外部チャネル（Slack/Email/LINE/Teams）へ同じ結果を配信する」処理が頻出する: 今回のNotificationChannel Protocol + RetryPolicy委譲 + TaskGroup並行実行という構造は、そのまま横展開できる汎用パターンとして再利用できる
• 「継承だけがコードの再利用手段ではない」という視点は、決済ゲートウェイ連携（クレジットカード/コンビニ決済/QRコード決済）のような「共通のインターフェースを満たすが、内部実装が大きく異なる」処理全般に応用できる

今日のまとめ

次のステップ

• 発展問題: RetryPolicyがリトライ対象外にすべきエラー（例: LINE Notifyのトークンが無効=401エラー、リトライしても永久に成功しない）を、例外の型で区別してリトライを打ち切れるように拡張せよ（第10章「例外の握り潰し・null問題」との関連）
• 発展問題: InventorySyncDispatcher.notify_allで一部チャネルだけ失敗した場合、ExceptionGroupから失敗したチャネルのみを特定し、そのチャネルだけ再送するリカバリー処理を設計せよ
• 参考: 「良いコード・悪いコードで学ぶ設計入門」第7章（継承より委譲、スマートUI・神クラス対処）、第6章（関心の分離）との関連、Python公式ドキュメント typing.Protocol、asyncio.TaskGroup

自己評価（あとで記入）