A+B 複合 — Pydantic v2 × OpenTelemetry × Secret Manager + Terraform 最小権限設計

📋 コピーimport requests
import json
import os

BQ_TABLE = "project-id.dataset.table"

def fetch_orders(api_url, token):
    r = requests.get(api_url, headers={"Authorization": "Bearer " + token})
    data = r.json()
    return data   # ← 問題: raise_for_status() なし

def transform(data):
    result = []
    for d in data:
        item = {}
        item["order_id"] = d["id"]           # ← 問題: フィールド欠損時に KeyError
        item["user_id"] = d["customer"]["id"] # ← Pydantic バリデーションなし
        item["total"] = d["amount"]
        item["status"] = d["state"]
        result.append(item)
    return result

def load_to_bq(rows, table):
    from google.cloud import bigquery
    client = bigquery.Client()
    errors = client.insert_rows_json(table, rows)  # ← 問題: streaming insert
    if errors:
        print("Error:", errors)  # ← 問題: sys.exit(1) なし

def main():
    token = os.environ["API_TOKEN"]
    url = "https://api.example.com/orders"
    raw = fetch_orders(url, token)
    rows = transform(raw)
    load_to_bq(rows, BQ_TABLE)
    print("done")  # ← 問題: logging を使っていない

📋 コピー"""
Cloud Run v2 ジョブ: 受注データ取得 → 変換 → BigQuery バッチロード
Design Patterns: Repository Pattern, Dependency Injection (tracer), Fail-Fast
"""
from __future__ import annotations

import logging
import os
import sys
import tempfile
import json
from typing import Final

import requests
from google.cloud import bigquery
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from pydantic import BaseModel, Field, ValidationError

# --- 定数 ---------------------------------------------------------------
API_URL: Final[str] = "https://api.example.com/orders"
BQ_TABLE: Final[str] = os.environ["BQ_TABLE"]
OTEL_ENDPOINT: Final[str] = os.environ.get("OTEL_ENDPOINT", "http://localhost:4317")

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
logger = logging.getLogger(__name__)


# --- スキーマ定義（Pydantic v2）------------------------------------------
class CustomerRef(BaseModel):
    id: str


class RawOrder(BaseModel):
    """外部 API レスポンスの 1 件分スキーマ"""
    id: str
    customer: CustomerRef
    amount: float
    state: str


class OrderRow(BaseModel):
    """BigQuery 書き込み用スキーマ"""
    order_id: str
    user_id: str
    total: float
    status: str


# --- OTel セットアップ ---------------------------------------------------
def setup_tracer() -> trace.Tracer:
    provider = TracerProvider()
    exporter = OTLPSpanExporter(endpoint=OTEL_ENDPOINT, insecure=True)
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    return trace.get_tracer(__name__)


# --- Repository ----------------------------------------------------------
def fetch_orders(api_url: str, token: str, tracer: trace.Tracer) -> list[RawOrder]:
    with tracer.start_as_current_span("fetch_orders") as span:
        response = requests.get(
            api_url,
            headers={"Authorization": f"Bearer {token}"},
            timeout=30,
        )
        response.raise_for_status()  # 4xx/5xx で例外を送出（Fail-Fast）

        raw_data: list[dict] = response.json()
        span.set_attribute("orders.count", len(raw_data))
        logger.info("Fetched %d orders", len(raw_data))

        orders: list[RawOrder] = []
        for item in raw_data:
            try:
                orders.append(RawOrder.model_validate(item))
            except ValidationError as exc:
                logger.warning("Skipping invalid order %s: %s", item.get("id"), exc)

        return orders


def transform(orders: list[RawOrder], tracer: trace.Tracer) -> list[OrderRow]:
    with tracer.start_as_current_span("transform"):
        return [
            OrderRow(
                order_id=o.id,
                user_id=o.customer.id,
                total=o.amount,
                status=o.state,
            )
            for o in orders
        ]


def load_to_bq(
    rows: list[OrderRow],
    table: str,
    client: bigquery.Client,
    tracer: trace.Tracer,
) -> None:
    with tracer.start_as_current_span("load_to_bq") as span:
        span.set_attribute("bq.table", table)
        span.set_attribute("bq.rows", len(rows))

        records = [r.model_dump() for r in rows]

        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as tmp:
            for record in records:
                tmp.write(json.dumps(record) + "\n")
            tmp_path = tmp.name

        with open(tmp_path, "rb") as f:
            job_config = bigquery.LoadJobConfig(
                source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON,
                write_disposition=bigquery.WriteDisposition.WRITE_APPEND,
                autodetect=False,
            )
            job = client.load_table_from_file(f, table, job_config=job_config)

        job.result()  # 完了まで待機

        if job.errors:
            raise RuntimeError(f"BigQuery load failed: {job.errors}")

        logger.info("Loaded %d rows to %s", len(rows), table)


# --- エントリーポイント --------------------------------------------------
def main() -> None:
    tracer = setup_tracer()

    with tracer.start_as_current_span("pipeline.main"):
        token = os.environ["API_TOKEN"]
        bq_client = bigquery.Client()  # 1 度だけ生成して DI

        try:
            raw_orders = fetch_orders(API_URL, token, tracer)
            order_rows = transform(raw_orders, tracer)
            load_to_bq(order_rows, BQ_TABLE, bq_client, tracer)
        except requests.HTTPError as exc:
            logger.error("API error: %s", exc)
            sys.exit(1)  # Argo Workflows が失敗として検知できるように非ゼロ終了
        except RuntimeError as exc:
            logger.error("BQ load error: %s", exc)
            sys.exit(1)

    logger.info("Pipeline completed successfully")


if __name__ == "__main__":
    main()