弱点補強 — GCP Pub/Sub × BigQuery × dbt イベント駆動パイプライン

📋 コピー# good_pipeline.py
from __future__ import annotations

import json
import logging
from datetime import UTC, datetime
from enum import StrEnum
from functools import cached_property

from google.cloud import bigquery, pubsub_v1
from opentelemetry import trace
from opentelemetry.trace import StatusCode
from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings

logger = logging.getLogger(__name__)
tracer = trace.get_tracer(__name__)


# ---- 設定 (環境変数から読み込み) ----

class Settings(BaseSettings):
    gcp_project_id: str
    pubsub_topic_id: str
    bq_dataset: str
    bq_table: str

    class Config:
        env_file = ".env"


_settings: Settings | None = None

def get_settings() -> Settings:
    global _settings
    if _settings is None:
        _settings = Settings()
    return _settings


# ---- ドメインモデル ----

class EventType(StrEnum):
    CART_ADD = "cart_add"
    PURCHASE = "purchase"
    CANCEL = "cancel"


class UserEvent(BaseModel):
    """購買行動イベント。Pydantic v2 で型・値バリデーション済み"""
    event_type: EventType
    user_id: str = Field(min_length=1, max_length=64, pattern=r"^[a-zA-Z0-9_-]+$")
    amount: float = Field(ge=0, le=10_000_000)
    campaign_id: str | None = Field(default=None, max_length=64)
    occurred_at: datetime = Field(default_factory=lambda: datetime.now(UTC))


# ---- インフラ層 (cached_property でシングルトン) ----

class PipelineClients:
    """BQ・Pub/Sub クライアントのシングルトン。Workload Identity で認証"""

    @cached_property
    def bq(self) -> bigquery.Client:
        return bigquery.Client(project=get_settings().gcp_project_id)

    @cached_property
    def publisher(self) -> pubsub_v1.PublisherClient:
        return pubsub_v1.PublisherClient()


_clients = PipelineClients()


# ---- パブリッシュ (Pub/Sub) ----

def publish_event(event: UserEvent) -> None:
    settings = get_settings()
    topic_path = _clients.publisher.topic_path(
        settings.gcp_project_id, settings.pubsub_topic_id
    )

    with tracer.start_as_current_span("pubsub.publish") as span:
        span.set_attribute("event.type", event.event_type)
        span.set_attribute("user.id", event.user_id)

        try:
            payload = event.model_dump_json().encode("utf-8")
            future = _clients.publisher.publish(topic_path, payload)
            message_id = future.result(timeout=10.0)
            span.set_attribute("pubsub.message_id", message_id)
        except Exception as exc:
            span.set_status(StatusCode.ERROR, str(exc))
            raise


# ---- BigQuery: パラメータ化クエリ ----

def get_campaign_roi(campaign_id: str, date: str) -> float | None:
    settings = get_settings()
    full_table = f"`{settings.gcp_project_id}.{settings.bq_dataset}.{settings.bq_table}`"

    query = f"""
        SELECT
            campaign_id,
            SUM(amount)  AS total_revenue,
            COUNT(*)     AS event_count,
            SAFE_DIVIDE(SUM(amount), COUNT(*)) AS avg_amount
        FROM {full_table}
        WHERE DATE(_PARTITIONTIME) = @target_date
          AND campaign_id = @campaign_id
          AND event_type = 'purchase'
        GROUP BY campaign_id
    """

    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ScalarQueryParameter("target_date", "DATE", date),
            bigquery.ScalarQueryParameter("campaign_id", "STRING", campaign_id),
        ]
    )

    with tracer.start_as_current_span("bigquery.query_campaign_roi") as span:
        span.set_attribute("campaign.id", campaign_id)
        try:
            rows = list(_clients.bq.query(query, job_config=job_config).result())
            if not rows:
                return None
            return rows[0].avg_amount
        except Exception as exc:
            span.set_status(StatusCode.ERROR, str(exc))
            raise


# ---- Cloud Run エントリーポイント ----

def process_webhook(request) -> tuple[str, int]:
    with tracer.start_as_current_span("webhook.process") as span:
        try:
            raw = request.get_json(silent=True)
            if raw is None:
                return "Bad Request: Invalid JSON", 400

            event = UserEvent.model_validate(raw)
            span.set_attribute("event.type", event.event_type)

            publish_event(event)
            return "Accepted", 202

        except ValueError as exc:
            return f"Unprocessable Entity: {exc}", 422
        except Exception as exc:
            span.set_status(StatusCode.ERROR, str(exc))
            return "Internal Server Error", 500