AB: コーディング × インフラ複合 — BigQuery → GCS エクスポートバッチ改善

📋 コピー"""
orders_export.py — BigQuery → GCS エクスポートバッチ

設計:
- 環境変数で設定値を注入（12-Factor App）
- ストリーミング書き込みでメモリ効率化
- 構造化ロギング（DataDog / Cloud Logging 対応）
- sys.exit(1) で Argo Workflows に失敗を通知
"""

import argparse
import json
import logging
import os
import sys
from datetime import date, datetime

from google.cloud import bigquery, storage

# ---------------------------------------------------------------------------
# 定数（環境変数で上書き可能）
# ---------------------------------------------------------------------------
BUCKET_NAME: str = os.environ["EXPORT_BUCKET"]          # 例: "prod-export-bucket"
PROJECT_ID: str  = os.environ["GCP_PROJECT"]            # 例: "my-project-prod"
DATASET_ID: str  = os.environ.get("BQ_DATASET", "ecom") # デフォルト値あり

EXPORT_BLOB_TEMPLATE = "orders/date={date}/orders.jsonl"

# ---------------------------------------------------------------------------
# ロガー設定
# ---------------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger(__name__)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="BigQuery → GCS orders exporter")
    parser.add_argument(
        "--date",
        type=lambda s: datetime.strptime(s, "%Y-%m-%d").date(),
        default=date.today(),
        help="エクスポート対象日付 (YYYY-MM-DD). デフォルト: 本日",
    )
    return parser.parse_args()


def build_query(target_date: date) -> str:
    return f"""
        SELECT order_id, customer_id, amount, status, created_at
        FROM `{PROJECT_ID}.{DATASET_ID}.orders`
        WHERE DATE(created_at) = @target_date
    """


def stream_export(target_date: date) -> int:
    bq_client  = bigquery.Client(project=PROJECT_ID)
    gcs_client = storage.Client(project=PROJECT_ID)

    query = build_query(target_date)
    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ScalarQueryParameter("target_date", "DATE", target_date.isoformat())
        ]
    )

    logger.info("Starting BigQuery query for date=%s", target_date)
    query_job = bq_client.query(query, job_config=job_config)

    blob_name = EXPORT_BLOB_TEMPLATE.format(date=target_date.isoformat())
    blob = gcs_client.bucket(BUCKET_NAME).blob(blob_name)

    row_count = 0
    # ストリーミング書き込み: 全行をメモリに乗せない
    with blob.open("w") as gcs_file:
        for row in query_job.result():
            gcs_file.write(json.dumps(dict(row), default=str) + "\n")
            row_count += 1
            if row_count % 10_000 == 0:
                logger.info("Streamed %d rows...", row_count)

    logger.info(
        "Export complete: date=%s rows=%d gs://%s/%s",
        target_date, row_count, BUCKET_NAME, blob_name,
    )
    return row_count


if __name__ == "__main__":
    args = parse_args()
    try:
        exported = stream_export(args.date)
    except Exception as exc:
        logger.error("Export failed: %s", exc, exc_info=True)
        sys.exit(1)  # Argo Workflows に非ゼロ終了コードで失敗を通知