A+B 複合 — asyncio バッチ設計 × GKE Gateway API + HPA

📋 コピー"""orders_upload.py — ECサイト注文データの BigQuery 一括アップロードバッチ。

良いコード・悪いコードで学ぶ設計入門
  - Ch4: 不変の活用（frozen dataclass, プロパティで読み取り専用）
  - Ch9: コレクションのカプセル化（OrderBatch が内部リストを隠蔽）
"""

from __future__ import annotations

import asyncio
import logging
from dataclasses import dataclass, field

from google.cloud import bigquery

# ── 定数 ────────────────────────────────────────────────
FETCH_CONCURRENCY: int = 20
ORDER_ID_RANGE: range = range(1, 101)
BQ_TABLE: str = "project.dataset.orders"

logger = logging.getLogger(__name__)


# ── コレクションの値オブジェクト（Ch9 + Ch4）────
@dataclass(slots=True)
class OrderBatch:
    """成功・失敗をカプセル化した注文バッチコレクション。

    外部から直接リストを操作させず、専用メソッド経由でのみ変更を受け付ける。
    取得時は tuple で返し、内部リストを書き換えられないようにする（Ch4 不変）。
    """

    _successes: list[dict] = field(default_factory=list, repr=False)
    _failures: list[tuple[int, str]] = field(default_factory=list, repr=False)

    def add_success(self, row: dict) -> None:
        self._successes.append(row)

    def add_failure(self, order_id: int, reason: str) -> None:
        self._failures.append((order_id, reason))

    @property
    def successes(self) -> tuple[dict, ...]:
        return tuple(self._successes)

    @property
    def failures(self) -> tuple[tuple[int, str], ...]:
        return tuple(self._failures)

    @property
    def success_count(self) -> int:
        return len(self._successes)

    @property
    def failure_count(self) -> int:
        return len(self._failures)


# ── フェッチ層（単一責任: データ取得のみ）──────────────────────
async def fetch_order(order_id: int) -> dict:
    await asyncio.sleep(0.01)
    if order_id % 7 == 0:
        raise ValueError(f"order {order_id} fetch failed")
    return {"id": order_id, "amount": order_id * 100, "status": "paid"}


async def fetch_all_orders(order_ids: range) -> OrderBatch:
    """指定した注文 ID 群を並列フェッチし OrderBatch にまとめる。"""
    sem = asyncio.Semaphore(FETCH_CONCURRENCY)

    async def _fetch_with_limit(order_id: int) -> dict | BaseException:
        async with sem:
            return await fetch_order(order_id)

    raw_results: list[dict | BaseException] = await asyncio.gather(
        *[_fetch_with_limit(oid) for oid in order_ids],
        return_exceptions=True,
    )

    batch = OrderBatch()
    for order_id, result in zip(order_ids, raw_results):
        if isinstance(result, BaseException):
            logger.warning("fetch failed order_id=%d reason=%s", order_id, result)
            batch.add_failure(order_id, str(result))
        else:
            batch.add_success(result)

    return batch


# ── アップロード層（単一責任: BQ 書き込みのみ）─────────────────
def upload_to_bigquery(batch: OrderBatch) -> None:
    if not batch.successes:
        logger.info("no rows to insert")
        return

    client = bigquery.Client()
    errors = client.insert_rows_json(BQ_TABLE, list(batch.successes))
    if errors:
        raise RuntimeError(f"BigQuery insert errors: {errors}")

    logger.info("inserted %d rows to %s", batch.success_count, BQ_TABLE)


# ── エントリーポイント ───────────────────────────────────────
async def main() -> None:
    batch = await fetch_all_orders(ORDER_ID_RANGE)
    upload_to_bigquery(batch)

    logger.info(
        "batch finished: success=%d, failed=%d",
        batch.success_count,
        batch.failure_count,
    )
    if batch.failures:
        logger.warning("failed order ids: %s", [fid for fid, _ in batch.failures])


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

📋 コピー# Gateway（外部 LB相当）
apiVersion: gateway.networking.k8s.io/v1
kind: Gateway
metadata:
  name: external-lb-gateway
  namespace: production
spec:
  gatewayClassName: gke-l7-global-external-managed
  listeners:
    - name: http
      port: 80
      protocol: HTTP
    - name: https
      port: 443
      protocol: HTTPS
      tls:
        mode: Terminate
        certificateRefs:
          - name: api-tls-cert  # Google-managed cert
---
# HTTP → HTTPS リダイレクト
apiVersion: gateway.networking.k8s.io/v1
kind: HTTPRoute
metadata:
  name: http-redirect
spec:
  parentRefs:
    - name: external-lb-gateway
      sectionName: http
  rules:
    - filters:
        - type: RequestRedirect
          requestRedirect:
            scheme: https
            statusCode: 301
---
# HTTPS ルーティング（カナリア対応）
apiVersion: gateway.networking.k8s.io/v1
kind: HTTPRoute
metadata:
  name: order-event-api
spec:
  parentRefs:
    - name: external-lb-gateway
      sectionName: https
  hostnames: ["api.example.com"]
  rules:
    - backendRefs:
        - name: order-event-api-stable
          port: 8080
          weight: 90
        - name: order-event-api-canary
          port: 8080
          weight: 10