弱点補強 — Pydantic v2 バリデーション設計 × BigQuery 書き込みパイプライン

📋 コピー"""campaign_event_processor.py — Pydantic v2 × BigQuery 書き込みパイプライン。

良いコード・悪いコードで学ぶ設計入門（改訂新版）
  - Ch2: 型の活用（Literal, Annotated+Field, AwareDatetime, 型ヒント全域）
  - Ch3: クラスによるカプセル化（EventProcessor 内部実装を隠蔽）
  - Ch4: 不変の活用（CampaignEvent frozen=True, ProcessResult frozen dataclass）
  - Ch10: エラー処理（BQInsertError raise, ValidationError ログ記録）
"""
from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import timezone
from typing import TYPE_CHECKING, Annotated, Final, Literal

from pydantic import AwareDatetime, BaseModel, Field, ValidationError

if TYPE_CHECKING:
    from google.cloud import bigquery

# ── ロガー ────────────────────────────────────────────────────────────────
logger = logging.getLogger(__name__)

# ── 定数（マジックストリング排除）──────────────────────────────────────────
MAX_DISCOUNT_RATE: Final[float] = 1.0  # 100% が上限
MIN_DISCOUNT_RATE: Final[float] = 0.0  # 0% は許可しない（gt=0.0）
ALLOWED_EVENT_TYPES: Final[frozenset[str]] = frozenset(
    {"impression", "click", "purchase"}
)


# ── Pydantic v2 モデル（Ch2: 型の活用）──────────────────────────────────────
class CampaignEvent(BaseModel):
    """販促キャンペーンイベントの入力スキーマ。

    Pydantic v2 の宣言的バリデーション機能を使い、
    型レベルで不正なデータを即検出する（Fail-Fast 原則 Ch10）。

    Attributes:
        event_id: イベントの一意識別子。
        campaign_id: キャンペーン識別子。
        user_id: ユーザー識別子。
        event_type: イベント種別（impression / click / purchase のみ許可）。
        discount_rate: 割引率（0.0 < rate <= 1.0）。
        occurred_at: イベント発生日時（UTC タイムゾーン付き）。
    """

    event_id: str
    campaign_id: str
    user_id: str
    # Literal で列挙制約を型レベルで表現（Ch2）
    # → "impressoin" などのタイポが ValidationError になる
    event_type: Literal["impression", "click", "purchase"]
    # Annotated + Field で範囲制約を型注釈に埋め込む（Ch2）
    # → gt=0.0（超過）, le=1.0（以下）: 0.0 < rate <= 1.0
    discount_rate: Annotated[float, Field(gt=MIN_DISCOUNT_RATE, le=MAX_DISCOUNT_RATE)]
    # AwareDatetime: Pydantic v2 組み込み型
    # → ISO 8601 文字列 → tzinfo 付き datetime に自動変換
    # → ナイーブ datetime（タイムゾーンなし）は ValidationError
    occurred_at: AwareDatetime

    model_config = {
        "frozen": True,       # Ch4: インスタンス不変（書き込み後の誤変更防止）
        "strict": False,      # str → datetime 等の型強制変換を許可
        "populate_by_name": True,
    }

    def to_bq_row(self) -> dict:
        """BigQuery insert_rows_json 用の dict に変換する。

        Returns:
            BQ に挿入可能な dict。occurred_at は UTC ISO 8601 文字列。
        """
        # UTC に正規化（AwareDatetime は timezone を保持しているが、
        # BQ の TIMESTAMP 型は UTC 前提なので明示的に変換）
        utc_dt = self.occurred_at.astimezone(timezone.utc)
        return {
            "event_id":      self.event_id,
            "campaign_id":   self.campaign_id,
            "user_id":       self.user_id,
            "event_type":    self.event_type,
            "discount_rate": self.discount_rate,
            "occurred_at":   utc_dt.isoformat(),  # "2026-05-31T12:00:00+00:00"
        }


# ── カスタム例外（Ch10: エラー処理）─────────────────────────────────────────
class BQInsertError(RuntimeError):
    """BigQuery insert_rows_json でエラーが発生した場合の例外。

    Argo Workflows のステップがこれをキャッチして
    failFast + Slack アラートを発行できる。
    """


# ── 処理結果値オブジェクト（Ch4: 不変の活用）───────────────────────────────
@dataclass(slots=True, frozen=True)  # frozen=True で処理結果の事後変更を防ぐ
class ProcessResult:
    """イベント処理結果を表す不変値オブジェクト。

    Attributes:
        inserted: BQ に正常挿入できた件数。
        failed: バリデーションエラーで処理できなかった件数。
        failed_rows: バリデーション失敗した元の raw dict（DLQ 用）。
    """

    inserted: int
    failed: int
    # tuple で不変コレクション（list を返さない — Ch7: コレクション操作の隠蔽）
    failed_rows: tuple[dict, ...] = field(default_factory=tuple)


# ── イベントプロセッサ（Ch3: クラスによるカプセル化）──────────────────────
class EventProcessor:
    """Pydantic v2 バリデーション → BigQuery 書き込みを担うプロセッサ。

    内部実装（BQ クライアント・テーブル名）を隠蔽し、
    `process()` メソッドのみ外部に公開する（Ch3）。

    Attributes:
        _bq_client: BigQuery クライアント（アンダースコアで隠蔽）。
        _table_id: 書き込み先テーブル（アンダースコアで隠蔽）。
    """

    # デフォルトテーブル ID（環境変数で上書き可）
    _DEFAULT_TABLE: Final[str] = "my-project.analytics.campaign_events"

    def __init__(
        self,
        bq_client: "bigquery.Client",
        table_id: str = _DEFAULT_TABLE,
    ) -> None:
        self._bq_client = bq_client  # アンダースコアで外部変更を防ぐ（Ch7）
        self._table_id = table_id

    def process(self, raw_events: list[dict]) -> ProcessResult:
        """raw イベント dict リストを検証し BigQuery に挿入する。

        バリデーション失敗行はスキップせず `ProcessResult.failed_rows` に記録する。
        BigQuery 書き込みエラーは `BQInsertError` として raise する（print しない）。

        Args:
            raw_events: 外部システム（Pub/Sub 等）から受け取った生データのリスト。

        Returns:
            挿入成功件数・失敗件数・失敗行を含む不変の ProcessResult。

        Raises:
            BQInsertError: BigQuery への挿入に失敗した場合。

        Example:
            >>> result = processor.process(events)
            >>> print(f"inserted={result.inserted}, failed={result.failed}")
            inserted=2, failed=1
        """
        valid_rows: list[dict] = []
        failed_rows: list[dict] = []

        for raw in raw_events:
            validated = self._validate(raw)
            if validated is not None:
                valid_rows.append(validated.to_bq_row())
            else:
                failed_rows.append(raw)  # 失敗行を保持（捨てない — DLQ 対応）

        if valid_rows:
            self._insert_to_bq(valid_rows)

        return ProcessResult(
            inserted=len(valid_rows),
            failed=len(failed_rows),
            failed_rows=tuple(failed_rows),  # list → tuple で不変化（Ch4）
        )

    def _validate(self, raw: dict) -> CampaignEvent | None:
        """単一 raw dict を CampaignEvent にバリデーションする。

        失敗時は None を返し、エラー詳細をログに記録する（raise しない）。

        Args:
            raw: バリデーション対象の raw dict。

        Returns:
            成功時は CampaignEvent インスタンス、失敗時は None。
        """
        try:
            return CampaignEvent.model_validate(raw)
        except ValidationError as exc:
            # ValidationError のフィールド別エラーをログに記録（print 禁止）
            logger.warning(
                "Validation failed",
                extra={
                    "raw_event_id": raw.get("event_id", "<unknown>"),
                    # include_url=False でドキュメント URL をログから除外
                    "errors": exc.errors(include_url=False),
                },
            )
            return None

    def _insert_to_bq(self, rows: list[dict]) -> None:
        """BigQuery に rows を挿入する。

        Args:
            rows: insert_rows_json に渡す dict のリスト。

        Raises:
            BQInsertError: BQ 側でエラーが返った場合（print しない）。
        """
        errors = self._bq_client.insert_rows_json(self._table_id, rows)
        if errors:
            # print ではなく例外として上位（Argo Workflows）に伝播させる（Ch10）
            raise BQInsertError(
                f"BigQuery insert failed for table {self._table_id!r}: {errors}"
            )
        logger.info("Inserted %d rows to BigQuery table %r", len(rows), self._table_id)