弱点補強 — asyncio TaskGroup × httpx 並列化 × OrderProcessingError 設計

📋 コピー"""order_processor.py — 注文イベント処理（asyncio TaskGroup × httpx）

設計の肝:
  - Ch2: 型の活用（dataclass, list[str], Final 定数）
  - asyncio TaskGroup（Python 3.11+）で独立API呼び出しを並列化
  - async with で httpx.AsyncClient を安全にライフサイクル管理
  - カスタム例外 OrderProcessingError で呼び出し元が失敗を検知可能
"""
from __future__ import annotations

import asyncio
import logging
from dataclasses import dataclass
from typing import Any, Final

import httpx

logger = logging.getLogger(__name__)

# ── 名前付き定数（URL・数値をハードコードしない）────────────────────────────
INVENTORY_API_BASE: Final[str] = "http://inventory-api"
POINTS_API_BASE:    Final[str] = "http://points-api"
NOTIFY_API_BASE:    Final[str] = "http://notify-api"

POINTS_PER_ITEM: Final[int] = 10      # 購入1アイテムあたりの付与ポイント
HTTP_TIMEOUT:    Final[float] = 10.0  # 全APIに共通タイムアウト（秒）


# ── カスタム例外 ──────────────────────────────────────────────────────────
class OrderProcessingError(RuntimeError):
    """注文処理のいずれかのステップが失敗した場合の例外。

    呼び出し元（Argo Workflows など）がこの例外をキャッチして
    リトライやアラートを発行できる。
    """


# ── 返り値の値オブジェクト（frozen=True で immutable、slots=True でメモリ効率化）
@dataclass(frozen=True, slots=True)
class OrderResult:
    """process_order の返り値。immutable で副作用を防ぐ（Ch3 値オブジェクト）。"""
    inventory: dict[str, Any]
    points:    dict[str, Any]
    notify:    dict[str, Any]


# ── 個別API呼び出し（単一責任：各関数が1エンドポイントのみ担当）────────────
async def check_inventory(
    client: httpx.AsyncClient, items: list[str]
) -> dict[str, Any]:
    """在庫APIを呼び出してアイテムの在庫状況を返す。

    Args:
        client: 共有の httpx.AsyncClient（ライフサイクルは呼び出し元が管理）。
        items: チェック対象のアイテムIDリスト。

    Returns:
        在庫APIのレスポンス JSON（dict）。

    Raises:
        OrderProcessingError: API呼び出しが失敗した場合。
    """
    try:
        items_param = ",".join(items)   # list を文字列にシリアライズ
        resp = await client.get(
            f"{INVENTORY_API_BASE}/check",
            params={"items": items_param},  # httpx が URL エンコードを担う
        )
        resp.raise_for_status()  # 4xx/5xx → httpx.HTTPStatusError を送出
    except httpx.HTTPError as exc:
        raise OrderProcessingError(f"Inventory check failed: {exc}") from exc
    logger.debug("Inventory check OK for %d items", len(items))
    return resp.json()


async def grant_points(
    client: httpx.AsyncClient, user_id: str, item_count: int
) -> dict[str, Any]:
    """ポイントAPIを呼び出してユーザーにポイントを付与する。

    Args:
        client: 共有の httpx.AsyncClient。
        user_id: ポイント付与対象のユーザーID。
        item_count: 購入アイテム数（ポイント計算の基数）。

    Returns:
        ポイントAPIのレスポンス JSON。

    Raises:
        OrderProcessingError: API呼び出しが失敗した場合。
    """
    points = item_count * POINTS_PER_ITEM  # 名前付き定数でマジックナンバーを排除
    try:
        resp = await client.post(
            f"{POINTS_API_BASE}/grant",
            json={"user_id": user_id, "points": points},
        )
        resp.raise_for_status()
    except httpx.HTTPError as exc:
        raise OrderProcessingError(
            f"Points grant failed for user {user_id}: {exc}"
        ) from exc
    logger.debug("Granted %d points to user %s", points, user_id)
    return resp.json()


async def send_notification(
    client: httpx.AsyncClient, user_id: str, order_id: str
) -> dict[str, Any]:
    """通知APIを呼び出して注文処理完了をユーザーへ通知する。

    Args:
        client: 共有の httpx.AsyncClient。
        user_id: 通知対象のユーザーID。
        order_id: 完了した注文ID（メッセージ本文に含める）。

    Returns:
        通知APIのレスポンス JSON。

    Raises:
        OrderProcessingError: API呼び出しが失敗した場合。
    """
    try:
        resp = await client.post(
            f"{NOTIFY_API_BASE}/send",
            json={"user_id": user_id, "message": f"Order {order_id} processed"},
        )
        resp.raise_for_status()
    except httpx.HTTPError as exc:
        raise OrderProcessingError(
            f"Notification failed for user {user_id}: {exc}"
        ) from exc
    logger.debug("Notification sent to user %s for order %s", user_id, order_id)
    return resp.json()


# ── メイン処理（3API を並列実行）─────────────────────────────────────────
async def process_order(
    order_id: str,
    user_id: str,
    items: list[str],
) -> OrderResult:
    """注文イベントを処理する。在庫チェック・ポイント付与・通知送信を並列実行する。

    Args:
        order_id: 処理する注文ID。
        user_id: 注文を行ったユーザーID。
        items: 注文に含まれるアイテムIDのリスト。

    Returns:
        各API呼び出しの結果を含む OrderResult（immutable）。

    Raises:
        OrderProcessingError: いずれかのAPI呼び出しが失敗した場合。

    Example:
        >>> result = asyncio.run(
        ...     process_order("ORD-123", "user-456", ["item-001", "item-002"])
        ... )
        >>> result.inventory["status"]
        'available'
    """
    # async with で AsyncClient を確実にクローズ（例外時も漏れなし）
    async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client:
        # TaskGroup（Python 3.11+）で3つの独立したAPI呼び出しを並列実行
        # gather() より TaskGroup の方が例外発生時に残タスクをキャンセルして安全
        async with asyncio.TaskGroup() as tg:
            t_inventory = tg.create_task(check_inventory(client, items))
            t_points    = tg.create_task(grant_points(client, user_id, len(items)))
            t_notify    = tg.create_task(send_notification(client, user_id, order_id))
        # with ブロックを抜けた時点で全タスクが完了（または例外が伝播）
        return OrderResult(
            inventory=t_inventory.result(),
            points=t_points.result(),
            notify=t_notify.result(),
        )