弱点補強 (A コーディング) — asyncio.TaskGroup × ContextVar × structlog（非同期メール一括送信 Bad→Good）

📋 コピー"""bulk_mail_sender.py — 非同期メール一括送信バッチ（Bad → Good リファクタリング）

良いコード・悪いコードで学ぶ設計入門（改訂新版）
  - Ch4: 不変の活用（frozen dataclass BulkSendResult × tuple で不変コレクション）
  - Ch7: コレクション操作の整理（_endpoint カプセル化 × errors list→tuple）
  - Ch10: エラー処理（MailSendError 集約・握りつぶし禁止・except*）
asyncio.TaskGroup / ContextVar / structlog の実践パターン（2026年時点デファクト）
"""
from __future__ import annotations

import asyncio
import time
import uuid
from contextvars import ContextVar
from dataclasses import dataclass, field
from typing import Final

import httpx
import structlog

# ── 名前付き定数（マジックナンバー根絶）────────────────────────────────────
MAX_CONCURRENT_SENDS: Final[int] = 20          # 同時送信の上限（rate limit 対策）
HTTP_TIMEOUT_SECONDS: Final[float] = 10.0      # HTTPタイムアウト（永久待機防止）
MAIL_API_ENDPOINT: Final[str] = "https://mail-api.internal/v1/send"

# ── ContextVar でリクエストID・キャンペーンIDを非同期コンテキストに伝搬 ────
# asyncio はタスク生成時にコンテキストをコピーするため、
# 親コルーチンで set() すれば全子コルーチンが get() で参照できる
# → スレッドセーフ・引数不要・コルーチンの署名がシンプルになる
REQUEST_ID_VAR: Final[ContextVar[str]] = ContextVar("request_id", default="")
CAMPAIGN_ID_VAR: Final[ContextVar[str]] = ContextVar("campaign_id", default="")


# ── カスタム例外（Ch10: エラー処理）─────────────────────────────────────────
class MailSendError(RuntimeError):
    """単一宛先へのメール送信失敗を表す例外。

    TaskGroup の ExceptionGroup からも型別捕捉できるよう
    RuntimeError のサブクラスとして定義する。

    Attributes:
        email:  送信先メールアドレス。
        reason: 失敗理由（HTTPステータスコードや "timeout" 等）。
    """

    def __init__(self, email: str, reason: str) -> None:
        super().__init__(f"mail_send_failed: {email!r} — {reason}")
        self.email = email
        self.reason = reason


# ── 送信結果値オブジェクト（Ch4: 不変の活用）──────────────────────────────
@dataclass(frozen=True, slots=True)
class BulkSendResult:
    """メール一括送信の処理結果を表す不変値オブジェクト。

    Attributes:
        sent:   送信成功件数。
        failed: 送信失敗件数。
        errors: 失敗した MailSendError のタプル（DLQ 用・不変コレクション）。
    """

    sent: int
    failed: int
    # list ではなく tuple で不変コレクション（Ch7: コレクション操作の整理）
    errors: tuple[MailSendError, ...] = field(default_factory=tuple)


# ── メール送信クラス ──────────────────────────────────────────────────────
class BulkMailSender:
    """非同期メール一括送信クラス。

    asyncio.TaskGroup で並列送信し、ContextVar でリクエストIDを伝搬、
    structlog で JSON 構造化ログを出力する。

    Attributes:
        _endpoint: メール送信 API エンドポイント（アンダースコアで隠蔽）。
    """

    def __init__(self, endpoint: str = MAIL_API_ENDPOINT) -> None:
        # アンダースコアでカプセル化（Ch7: コレクション操作の隠蔽）
        self._endpoint = endpoint

    async def send_bulk(
        self,
        campaign_id: str,
        recipients: list[dict],
    ) -> BulkSendResult:
        """recipients 全員に asyncio.TaskGroup で並列メール送信する。

        ContextVar にリクエストID・キャンペーンIDをセットすることで
        全サブコルーチンが引数なしで参照できる。

        Args:
            campaign_id: 送信対象キャンペーンID（ContextVar に伝搬）。
            recipients:  送信先リスト（各 dict に "email" / "name" キーを持つ）。

        Returns:
            送信成功件数・失敗件数・エラー詳細を含む BulkSendResult。
        """
        if not recipients:
            return BulkSendResult(sent=0, failed=0)

        # ContextVar にセット（全サブコルーチンが get() で参照できる）
        run_id = str(uuid.uuid4())
        REQUEST_ID_VAR.set(run_id)      # → 子コルーチンで REQUEST_ID_VAR.get()
        CAMPAIGN_ID_VAR.set(campaign_id)

        log = structlog.get_logger().bind(
            request_id=run_id,
            campaign_id=campaign_id,
            total=len(recipients),
        )
        log.info("bulk_mail_started")

        errors: list[MailSendError] = []
        # Semaphore で同時送信数を MAX_CONCURRENT_SENDS に制限（rate limit 対策）
        sem = asyncio.Semaphore(MAX_CONCURRENT_SENDS)

        # httpx.AsyncClient を TaskGroup の外で生成して接続プールを再利用
        async with httpx.AsyncClient(timeout=HTTP_TIMEOUT_SECONDS) as client:
            try:
                # TaskGroup: 全タスクを並列起動し、全完了を待つ
                async with asyncio.TaskGroup() as tg:
                    for r in recipients:
                        tg.create_task(self._send_one(client, sem, r))
            except* MailSendError as eg:
                # except*（Python 3.11+）: ExceptionGroup から型別に捕捉
                # eg.exceptions は失敗した全 MailSendError のリスト
                for exc in eg.exceptions:
                    errors.append(exc)

        sent_count = len(recipients) - len(errors)
        log.info("bulk_mail_finished", sent=sent_count, failed=len(errors))

        return BulkSendResult(
            sent=sent_count,
            failed=len(errors),
            errors=tuple(errors),  # list → tuple で不変コレクション（Ch7）
        )

    async def _send_one(
        self,
        client: httpx.AsyncClient,
        sem: asyncio.Semaphore,
        recipient: dict,
    ) -> None:
        """単一宛先へのメール送信。

        ContextVar からリクエストID・キャンペーンIDを取得して
        structlog にバインドすることで、全ログに追跡IDが自動付与される。

        Args:
            client:    共有 httpx.AsyncClient（接続プールを再利用）。
            sem:       同時送信数制限用 Semaphore（MAX_CONCURRENT_SENDS）。
            recipient: 送信先 dict（"email" / "name" キーを持つ）。

        Raises:
            MailSendError: 送信失敗（HTTP エラー・タイムアウト等）。
        """
        email = recipient.get("email", "")
        # ContextVar から参照（引数として渡す必要がない）
        log = structlog.get_logger().bind(
            request_id=REQUEST_ID_VAR.get(),   # 親コルーチンが set() した値
            campaign_id=CAMPAIGN_ID_VAR.get(),
            email=email,
        )

        async with sem:  # 同時実行数を MAX_CONCURRENT_SENDS に制限
            start = time.monotonic()
            try:
                resp = await client.post(
                    self._endpoint,
                    json={"to": email, "name": recipient.get("name", "")},
                )
                resp.raise_for_status()  # 4xx/5xx → httpx.HTTPStatusError
                elapsed_ms = int((time.monotonic() - start) * 1000)
                # structlog: JSON 構造化ログ（DataDog @elapsed_ms でフィルタ可）
                log.info("mail_sent", elapsed_ms=elapsed_ms, status=resp.status_code)

            except httpx.HTTPStatusError as exc:
                log.error("mail_failed", reason=f"HTTP {exc.response.status_code}")
                # 握りつぶし禁止: 呼び出し元（send_bulk）が except* で集約（Ch10）
                raise MailSendError(email, f"HTTP {exc.response.status_code}") from exc

            except httpx.TimeoutException:
                log.error("mail_failed", reason="timeout")
                raise MailSendError(email, "timeout")