A コーディング — asyncio.Queue Bounded Buffer × asyncio.TaskGroup 構造化並行性 × contextvars.ContextVar 暗黙伝播 × PEP 695 type 文 × frozen dataclass slots=True DispatchResult（キャンペーン配信バッチ バックプレッシャー Bad→Good Ch2/Ch4/Ch7/Ch10/Ch11）

📋 コピー"""campaign_dispatcher_async.py

Ch2:  PEP 695 type JobItem / StrEnum JobStatus / 名前付き定数
Ch4:  frozen dataclass slots=True DispatchResult（値オブジェクト）
Ch7:  MAX_QUEUE_SIZE / NUM_WORKERS / SENTINEL 名前付き定数
Ch10: contextvars.ContextVar 暗黙伝播 / Fail-Fast
Ch11: asyncio.Queue Bounded Buffer / TaskGroup 構造化並行性 / reduce 集計
"""
from __future__ import annotations
import asyncio, contextvars, functools, logging, random
from dataclasses import dataclass
from enum import StrEnum
from typing import Final

logger = logging.getLogger(__name__)

# ── PEP 695 型エイリアス（Ch2）
type JobItem = tuple[str, str]          # (recipient_id, message)
type WorkerResults = list[DispatchResult]

# ── 名前付き定数（Ch7）
MAX_QUEUE_SIZE: Final[int]       = 100   # バックプレッシャー閾値
NUM_WORKERS: Final[int]          = 5     # 並列ワーカー数
SUCCESS_RATE_THRESHOLD: Final[float] = 0.90
SENTINEL: Final[str]             = "__DONE__"  # ワーカー終了シグナル

# ── ContextVar（Ch10）
_CAMPAIGN_CTX: contextvars.ContextVar[str] = contextvars.ContextVar(
    "campaign_id", default=""
)

# ── StrEnum（Ch2）
class JobStatus(StrEnum):
    SUCCESS = "success"
    FAILED  = "failed"
    SKIPPED = "skipped"

# ── 値オブジェクト（Ch4）
@dataclass(frozen=True, slots=True)
class DispatchResult:
    """配信ジョブの実行結果（イミュータブル値オブジェクト）。"""
    sent_count: int
    success_count: int
    failed_ids: tuple[str, ...]  # tuple: hashable + immutable

    @property
    def success_rate(self) -> float:
        return self.success_count / self.sent_count if self.sent_count else 0.0

    @property
    def failure_count(self) -> int:
        return self.sent_count - self.success_count

    def is_healthy(self) -> bool:
        return self.success_rate >= SUCCESS_RATE_THRESHOLD

# ── プロデューサー
async def producer(queue: asyncio.Queue[JobItem], items: list[JobItem]) -> None:
    """ジョブを asyncio.Queue に投入する。maxsize 到達時に自動ブロック。"""
    campaign_id = _CAMPAIGN_CTX.get()  # ContextVar から取得
    logger.info("[Producer] campaign=%s total_jobs=%d", campaign_id, len(items))
    for item in items:
        await queue.put(item)          # バックプレッシャー: 満杯時に自動ブロック
    for _ in range(NUM_WORKERS):
        await queue.put((SENTINEL, ""))  # 終了シグナル

# ── ワーカー
async def worker(
    worker_id: int,
    queue: asyncio.Queue[JobItem],
    results: WorkerResults,
) -> None:
    """キューからジョブを取り出して配信する非同期ワーカー。"""
    campaign_id = _CAMPAIGN_CTX.get()  # ContextVar から暗黙取得
    sent, failed = 0, []
    while True:
        recipient_id, message = await queue.get()
        if recipient_id == SENTINEL:
            queue.task_done()
            break
        sent += 1
        try:
            await _send_stub(campaign_id, recipient_id, message)
        except RuntimeError as exc:
            logger.warning("[W%d] fail recipient=%s err=%s", worker_id, recipient_id, exc)
            failed.append(recipient_id)
        finally:
            queue.task_done()
    results.append(DispatchResult(
        sent_count=sent,
        success_count=sent - len(failed),
        failed_ids=tuple(failed),  # list → tuple でイミュータブル化
    ))

async def _send_stub(campaign_id: str, recipient_id: str, message: str) -> None:
    """外部送信 API の stub（95% 成功 / 5% 失敗）。"""
    await asyncio.sleep(0)
    if random.random() < 0.05:
        raise RuntimeError(f"送信失敗: {recipient_id}")

# ── 集計（functools.reduce）
def aggregate_results(partial: WorkerResults) -> DispatchResult:
    def merge(acc: DispatchResult, r: DispatchResult) -> DispatchResult:
        return DispatchResult(
            sent_count=acc.sent_count + r.sent_count,
            success_count=acc.success_count + r.success_count,
            failed_ids=acc.failed_ids + r.failed_ids,
        )
    return functools.reduce(merge, partial, DispatchResult(0, 0, ()))

# ── エントリポイント
async def run_campaign_async(campaign_id: str, jobs: list[JobItem]) -> DispatchResult:
    """非同期バックプレッシャー付きでキャンペーン配信を実行する。"""
    token = _CAMPAIGN_CTX.set(campaign_id)  # ContextVar をセット
    try:
        queue: asyncio.Queue[JobItem] = asyncio.Queue(maxsize=MAX_QUEUE_SIZE)
        partial_results: WorkerResults = []
        async with asyncio.TaskGroup() as tg:  # 構造化並行性
            tg.create_task(producer(queue, jobs))
            for wid in range(NUM_WORKERS):
                tg.create_task(worker(wid, queue, partial_results))
        result = aggregate_results(partial_results)
        if not result.is_healthy():
            logger.warning("campaign=%s rate=%.1f%% below threshold", campaign_id, result.success_rate * 100)
        return result
    finally:
        _CAMPAIGN_CTX.reset(token)  # ContextVar を確実にリセット