A+B 複合 — ファーストクラスコレクション × StrEnum × match × frozen dataclass(クーポンスタッキング Bad→Good Ch3/Ch4/Ch8/Ch9/Ch10)× Cloud Run Jobs × Cloud Tasks × Secret Manager rotation × Scheduler OIDC(Terraform)

2026-07-11 (Day 94) 土曜複合問題 ★★★★☆ Python 3.12 / StrEnum / match / frozen dataclass / ファーストクラスコレクション Cloud Run Jobs / Cloud Tasks / Secret Manager rotation / Cloud Scheduler OIDC

概要

📦

ファーストクラスコレクションで「ルールの集合」を隠蔽する

生の list を公開すると外部から自由に append/clear され不整合な状態を作れる。CouponBasket のように内部の tuple を隠蔽し、操作を専用メソッド(add_coupon)経由に限定することで、状態変化の経路を1箇所に集約できる。

🧊

不変な add_coupon で状態不整合を構造的に防ぐ

add_couponself を変更せず新しい CouponBasket を返すことで、あるコードパスの副作用が別のコードパスの前提を壊すバグを防げる。frozen=True, slots=True の値オブジェクトと組み合わせることでメモリ効率も両立する。

🔀

StrEnum × match で分岐とマジックストリングを一元化

"percent" のような文字列リテラルが各所に散らばると typo が実行時まで検出できない。StrEnum で定義し尽くし match 文で分岐すると、新しい種別追加時に case _: で未知値をフェイルラウドにでき、条件判定のネストも解消できる。

⏱️

Cloud Run Jobs × Cloud Tasks × Secret Manager rotation

常時起動 Service で動かしていたバッチを Cloud Run Jobs に移行して待機コストを排除し、Cloud Tasks のレート制御で外部APIの429エラーを構造的に防ぎ、Secret Manager rotation + Pub/Sub通知でAPIキーの平文管理から脱却する。

問題 A: コーディング — ファーストクラスコレクション × StrEnum × match(クーポンスタッキング Bad→Good)

以下の「悪いコード」は、ECサイト MOps チームのカート画面で複数クーポンを重ね掛け(スタッキング)する割引計算ロジックです。問題点を全て洗い出し、ファーストクラスコレクション・StrEnum・match 文・frozen dataclass(slots=True)・完全コンストラクタ を使って Bad→Good にリファクタリングしてください。

制約・前提条件

  • Python 3.12+(StrEnumenum.StrEnummatch 文は Python 3.10+)
  • CouponType / ConditionTypeStrEnum で定義し、マジックストリングを排除すること
  • Coupon / Condition / Cart / CartItem@dataclass(frozen=True, slots=True) の値オブジェクトとして実装すること
  • CouponBasket はファーストクラスコレクション(_coupons: tuple[Coupon, ...] を内部に隠蔽し、外部に生の list を公開しない)とし、add_coupon は新しい CouponBasket を返す不変設計にすること
  • Condition.is_satisfiedmatch 文で条件種別ごとに判定し、ネストループを解消すること
  • 割引合計は cart.total を超えないようクランプすること(マイナス残高防止)
  • Google スタイル docstring・インラインコメント・名前付き定数を含めること
期待する回答形式: 問題点の列挙(番号付き)+ 改善後コード + 実行例(input→output)+ 適用した設計パターン名と書籍対応章

悪いコード (Before) — カテゴリ A

このコードには 7つの設計上の問題 が隠れています。見つけてみてください。
bad_coupon_basket.py — public list・dict表現・3重ネストループ・上限チェックなし
class CouponBasket:
    def __init__(self):
        self.coupons = []  # 問題1: public list、外部から直接 mutate 可能

    def add_coupon(self, coupon):
        self.coupons.append(coupon)  # 問題7: 型チェック・重複IDチェックなし


def calculate_total_discount(basket, cart_total, cart_items):
    total_discount = 0
    applied_types = []
    for coupon in basket.coupons:                       # 問題3: 3重ネストループの起点
        ok = True
        for condition in coupon["conditions"]:           # ネスト2段目
            if condition["type"] == "min_amount":        # 問題6: マジックストリング
                if cart_total < condition["value"]:
                    ok = False
            if condition["type"] == "category":
                found = False
                for item in cart_items:                  # ネスト3段目
                    if item["category"] == condition["value"]:
                        found = True
                if not found:
                    ok = False
        if not ok:
            continue
        if coupon["type"] in applied_types:
            continue
        if coupon["type"] == "percent":                  # 問題6: マジックストリング
            total_discount += cart_total * coupon["rate"]
        elif coupon["type"] == "fixed":
            total_discount += coupon["amount"]
        applied_types.append(coupon["type"])
    # 問題4: 上限チェックなし。合算がcart_totalを超えるとマイナス残高になり得る
    return total_discount
問題点サマリー(7点)
1CouponBasket.coupons が public list(Ch9) — 外部から basket.coupons.append(...) のように直接 mutate でき不整合な状態を作れる。ファーストクラスコレクションとして隠蔽する
2Coupon/Condition が dict 表現(Ch2)coupon["rate"] のようなキーアクセスは typo しても実行時まで検出されない。frozen dataclass の値オブジェクトにする
33重ネストループ(Ch9) — coupon × condition × cart_items が入れ子で可読性・計算量が劣化している。条件判定を委譲し1段に平坦化する
4割引合計の上限チェックなし(Ch10) — クーポンを積み上げると cart_total を超え、マイナス残高になり得る
5重複適用防止が緩い(Ch10) — 種別単位のチェックのみで、同一クーポンIDの二重登録を検出できない
6マジックストリングの散在(Ch10)"percent", "fixed", "min_amount", "category" がリテラルとして各所に書かれている
7add_coupon にバリデーションがない(Ch3) — 型チェックも重複チェックもなく任意のオブジェクトを追加できる

ヒント A(段階的開示)

ヒント1 — 方向性
CouponBasket は「クーポンのリスト」という概念そのものをファーストクラスコレクションとしてクラス化する。外部からは list として触らせず、add_coupon は既存の basket を変更せず新しい CouponBasket を返す(不変)。条件判定(min_amount / category)は Condition.is_satisfied(cart) に委譲し、呼び出し側は all(...) で早期 continue すればネストは1段で済む。
ヒント2 — アプローチ
  • class CouponType(StrEnum): PERCENT = "percent"; FIXED = "fixed"
  • class ConditionType(StrEnum): MIN_AMOUNT = "min_amount"; CATEGORY = "category"
  • @dataclass(frozen=True, slots=True) class Conditionis_satisfied(self, cart: Cart) -> boolmatch self.type: で実装
  • @dataclass(frozen=True, slots=True) class Coupon__post_init__PERCENT なら rate 必須、FIXED なら amount 必須をバリデーション(完全コンストラクタ)
  • CouponBasket_coupons: tuple[Coupon, ...] を隠蔽し add_coupon が新インスタンスを返す。__iter__/__len__ で読み取り専用アクセスのみ許可
  • calculate_total_discountmin(total_discount, cart.total) で上限クランプ
ヒント3 — コードの骨格
from __future__ import annotations
from dataclasses import dataclass, field
from enum import StrEnum

class CouponType(StrEnum):
    PERCENT = "percent"
    FIXED = "fixed"

class ConditionType(StrEnum):
    MIN_AMOUNT = "min_amount"
    CATEGORY = "category"

@dataclass(frozen=True, slots=True)
class Condition:
    type: ConditionType
    value: str | int

    def is_satisfied(self, cart: "Cart") -> bool:
        match self.type:
            case ConditionType.MIN_AMOUNT:
                return cart.total >= int(self.value)
            case ConditionType.CATEGORY:
                return any(item.category == self.value for item in cart.items)
        raise AssertionError(f"未知の条件種別: {self.type}")

@dataclass(frozen=True, slots=True)
class CouponBasket:
    _coupons: tuple["Coupon", ...] = field(default_factory=tuple)

    def add_coupon(self, coupon: "Coupon") -> "CouponBasket":
        ...  # 重複IDチェック → 新インスタンスを返す

    def __iter__(self):
        return iter(self._coupons)

問題点分析 — カテゴリ A

#問題点分類改善方法
1CouponBasket.coupons が public listカプセル化 Ch9ファーストクラスコレクションで隠蔽
2Coupon/Condition が dict 表現型の活用 Ch2frozen dataclass の値オブジェクト化
33重ネストループコレクション Ch9条件判定の委譲 + all() で平坦化
4割引合計の上限チェックなし設計の悪魔 Ch10min(total_discount, cart.total) でクランプ
5重複適用防止が緩い設計の悪魔 Ch10add_coupon 側でID重複を検出
6マジックストリングの散在設計の悪魔 Ch10StrEnum で定義し尽くす
7add_coupon にバリデーションなし完全コンストラクタ Ch3不変な add_coupon + 型検証

模範解答 A

Before — public list・dict表現・3重ネスト・上限チェックなし
class CouponBasket:
    def __init__(self):
        self.coupons = []  # public list

    def add_coupon(self, coupon):
        self.coupons.append(coupon)  # バリデーションなし

def calculate_total_discount(basket, cart_total, cart_items):
    total_discount = 0
    for coupon in basket.coupons:
        for condition in coupon["conditions"]:      # ネスト2段目
            if condition["type"] == "category":
                for item in cart_items:              # ネスト3段目
                    ...
        if coupon["type"] == "percent":              # マジックストリング
            total_discount += cart_total * coupon["rate"]
    return total_discount   # 上限チェックなし
After — ファーストクラスコレクション × StrEnum × match × frozen dataclass
"""coupon_basket.py — ファーストクラスコレクション(Ch3/Ch4/Ch8/Ch9/Ch10)"""
from __future__ import annotations

from dataclasses import dataclass, field
from enum import StrEnum
from typing import Final

MAX_COUPONS_PER_TYPE: Final[int] = 1  # 名前付き定数


class CouponType(StrEnum):
    PERCENT = "percent"
    FIXED = "fixed"


class ConditionType(StrEnum):
    MIN_AMOUNT = "min_amount"
    CATEGORY = "category"


@dataclass(frozen=True, slots=True)
class CartItem:
    category: str


@dataclass(frozen=True, slots=True)
class Cart:
    total: int
    items: tuple[CartItem, ...] = field(default_factory=tuple)


@dataclass(frozen=True, slots=True)
class Condition:
    """クーポン適用条件(値オブジェクト)。"""
    type: ConditionType
    value: str | int

    def is_satisfied(self, cart: Cart) -> bool:
        """match による分岐の一元化(Ch8)。"""
        match self.type:
            case ConditionType.MIN_AMOUNT:
                return cart.total >= int(self.value)
            case ConditionType.CATEGORY:
                return any(i.category == self.value for i in cart.items)
        raise AssertionError(f"未知の条件種別: {self.type}")  # 握りつぶし禁止


@dataclass(frozen=True, slots=True)
class Coupon:
    """クーポン(完全コンストラクタ)。"""
    id: str
    type: CouponType
    conditions: tuple[Condition, ...] = field(default_factory=tuple)
    rate: float | None = None
    amount: int | None = None

    def __post_init__(self) -> None:
        if self.type is CouponType.PERCENT and self.rate is None:
            raise ValueError(f"PERCENT には rate が必須: id={self.id}")
        if self.type is CouponType.FIXED and self.amount is None:
            raise ValueError(f"FIXED には amount が必須: id={self.id}")

    def discount_amount(self, cart: Cart) -> int:
        match self.type:
            case CouponType.PERCENT:
                return int(cart.total * self.rate)
            case CouponType.FIXED:
                return self.amount
        raise AssertionError(f"未知のクーポン種別: {self.type}")


@dataclass(frozen=True, slots=True)
class CouponBasket:
    """クーポンのファーストクラスコレクション(不変・カプセル化)。"""
    _coupons: tuple[Coupon, ...] = field(default_factory=tuple)

    def add_coupon(self, coupon: Coupon) -> CouponBasket:
        """新しい CouponBasket を返す(自分自身は変更しない)。"""
        if any(c.id == coupon.id for c in self._coupons):
            raise ValueError(f"クーポンID重複: {coupon.id}")
        return CouponBasket(_coupons=(*self._coupons, coupon))

    def __iter__(self):
        return iter(self._coupons)   # 読み取り専用の反復のみ許可

    def __len__(self) -> int:
        return len(self._coupons)


def calculate_total_discount(basket: CouponBasket, cart: Cart) -> int:
    """クーポンをスタッキング適用し、合計割引額を計算する。"""
    total_discount = 0
    applied_types: set[CouponType] = set()
    for coupon in basket:  # ネストなしの単一ループ(Ch9)
        if coupon.type in applied_types and len(applied_types) >= MAX_COUPONS_PER_TYPE:
            continue
        if not all(c.is_satisfied(cart) for c in coupon.conditions):
            continue  # 早期 continue(Ch8)
        total_discount += coupon.discount_amount(cart)
        applied_types.add(coupon.type)
    return min(total_discount, cart.total)  # マイナス残高防止(Ch10)
cart = Cart(
    total=12000,
    items=(CartItem(category="books"), CartItem(category="electronics")),
)

basket = CouponBasket()
basket = basket.add_coupon(Coupon(
    id="C1", type=CouponType.PERCENT, rate=0.1,
    conditions=(Condition(type=ConditionType.MIN_AMOUNT, value=10000),),
))
basket = basket.add_coupon(Coupon(
    id="C2", type=CouponType.FIXED, amount=5000,
    conditions=(Condition(type=ConditionType.CATEGORY, value="electronics"),),
))

print(calculate_total_discount(basket, cart))
# → 6200  (12000*0.1=1200 + 5000 = 6200、cart.total=12000未満なのでクランプなし)

# 重複ID登録は例外
basket.add_coupon(Coupon(id="C1", type=CouponType.FIXED, amount=1000))
# → ValueError: クーポンID重複: C1

# 元の basket は不変(add_coupon は新インスタンスを返す)
print(len(basket))  # → 2

# 割引が cart.total を超えるケース(クランプ確認)
huge_cart = Cart(total=3000, items=(CartItem(category="electronics"),))
huge_basket = CouponBasket().add_coupon(
    Coupon(id="C3", type=CouponType.FIXED, amount=8000)
)
print(calculate_total_discount(huge_basket, huge_cart))
# → 3000(8000ではなくcart.totalでクランプされる)
ポイント適用した設計原則/パターン書籍対応章
CouponBasket の _coupons 隠蔽ファーストクラスコレクションCh9
add_coupon が新インスタンスを返す不変・副作用排除Ch4
Coupon.__post_init__ バリデーション完全コンストラクタCh3
Condition.is_satisfied への委譲ループネスト解消・関心の分離Ch8/Ch9
StrEnum でマジックストリング排除設計の悪魔対策Ch10
min(total_discount, cart.total)境界値のガードCh10
raise AssertionError で未知値検出例外の握りつぶし禁止Ch10
# tests/test_coupon_basket.py
import pytest
from coupon_basket import (
    Cart, CartItem, Condition, ConditionType, Coupon, CouponType,
    CouponBasket, calculate_total_discount,
)


class TestCouponBasket:
    def test_add_coupon_returns_new_instance(self):
        basket = CouponBasket()
        new_basket = basket.add_coupon(Coupon(id="C1", type=CouponType.FIXED, amount=100))
        assert len(basket) == 0        # 元のインスタンスは不変
        assert len(new_basket) == 1

    def test_duplicate_id_raises(self):
        basket = CouponBasket().add_coupon(Coupon(id="C1", type=CouponType.FIXED, amount=100))
        with pytest.raises(ValueError, match="クーポンID重複"):
            basket.add_coupon(Coupon(id="C1", type=CouponType.FIXED, amount=200))

    def test_iteration_read_only(self):
        basket = CouponBasket().add_coupon(Coupon(id="C1", type=CouponType.FIXED, amount=100))
        assert [c.id for c in basket] == ["C1"]


class TestCoupon:
    def test_percent_requires_rate(self):
        with pytest.raises(ValueError, match="rate が必須"):
            Coupon(id="C1", type=CouponType.PERCENT)

    def test_fixed_requires_amount(self):
        with pytest.raises(ValueError, match="amount が必須"):
            Coupon(id="C1", type=CouponType.FIXED)

    def test_frozen_immutability(self):
        c = Coupon(id="C1", type=CouponType.FIXED, amount=100)
        with pytest.raises(Exception):
            c.amount = 200  # type: ignore[misc]


class TestCalculateTotalDiscount:
    def test_stacking_percent_and_fixed(self):
        cart = Cart(total=12000, items=(CartItem(category="electronics"),))
        basket = (
            CouponBasket()
            .add_coupon(Coupon(id="C1", type=CouponType.PERCENT, rate=0.1,
                                conditions=(Condition(ConditionType.MIN_AMOUNT, 10000),)))
            .add_coupon(Coupon(id="C2", type=CouponType.FIXED, amount=5000,
                                conditions=(Condition(ConditionType.CATEGORY, "electronics"),)))
        )
        assert calculate_total_discount(basket, cart) == 6200

    def test_clamped_to_cart_total(self):
        cart = Cart(total=3000, items=(CartItem(category="electronics"),))
        basket = CouponBasket().add_coupon(Coupon(id="C1", type=CouponType.FIXED, amount=8000))
        assert calculate_total_discount(basket, cart) == 3000

    def test_condition_not_satisfied_excludes_coupon(self):
        cart = Cart(total=5000, items=(CartItem(category="books"),))
        basket = CouponBasket().add_coupon(
            Coupon(id="C1", type=CouponType.PERCENT, rate=0.1,
                   conditions=(Condition(ConditionType.MIN_AMOUNT, 10000),))
        )
        assert calculate_total_discount(basket, cart) == 0

問題 B: インフラ — Cloud Run Jobs × Cloud Tasks × Secret Manager 自動ローテーション(Terraform)

ECサイト MOps チームでは、深夜バッチでベンダー(卸元)APIから商品マスタ・クーポン原資データを同期する処理が、以下の構成で運用されており課題を抱えています。

現状の課題:
  • Cloud Run Servicemin-instances=1 で常時起動)を Cloud Scheduler の HTTP トリガーで叩いている → 実行時間は深夜の数分間だけなのに24時間課金が発生している
  • ベンダー API キーが Cloud Run revision の環境変数に平文で設定されている(gcloud run services describe で誰でも閲覧可能)→ ローテーションの仕組みもない
  • ハンドラー内で約5,000 SKU 分を for ループで同時に無制限リクエスト → ベンダー API のレート制限に頻繁に抵触し429エラーで同期が不完全に終わる
  • 実行用サービスアカウントがデフォルトSA(プロジェクト編集者相当の過剰権限)のまま

要件

#要件
1google_cloud_run_v2_job(Cloud Run Jobs)に移行し、実行時間のみ課金される構成にすること
2google_cloud_tasks_queue でベンダー API 呼び出しをキューイングし、rate_limits でレート制御、retry_config で指数バックオフを設定すること
3ベンダー API キーを google_secret_manager_secret に格納し、rotation + Pub/Sub 通知でローテーション運用を組み込むこと。Job は専用SAで secretAccessor のみ付与すること
4Cloud Scheduler は OIDC 認証の専用サービスアカウントで Cloud Run Jobs Execution API を起動すること(IAM 最小権限)

ヒント B(段階的開示)

ヒント1 — 方向性
Cloud Run Jobs は「起動して終了する」バッチ専用リソースで、Service と異なり min-instances の概念がなく実行時間のみ課金される。Secret Manager の rotation はキー自体を自動生成しない(あくまで「そろそろ更新してください」という Pub/Sub 通知機能)ため、通知を受けて新キーを取得・格納する運用フロー(別 Cloud Function 等)とセットで設計する。Cloud Tasks は「エンキュー→ワーカーが一定レートで処理」というバックプレッシャーの標準パターン。
ヒント2 — Terraform リソース構成
  • google_cloud_run_v2_jobtemplate.template.containers の入れ子構造に注意)
  • google_cloud_tasks_queuerate_limits + retry_config
  • google_secret_manager_secretrotation.rotation_period + topics)+ google_secret_manager_secret_iam_member
  • google_cloud_scheduler_jobhttp_target.oauth_token.service_account_email で OIDC 起動)
  • Job用SAとScheduler用SAを分離し、それぞれ最小権限のロールのみ付与する
ヒント3 — リソースの骨格
Cloud Run Jobs + Cloud Tasks
resource "google_cloud_run_v2_job" "vendor_sync_job" {
  name     = "vendor-sync-job"
  location = var.region
  template {
    task_count  = 1
    parallelism = 1
    template {
      service_account = google_service_account.vendor_sync_job_sa.email
      containers { image = "..." }
    }
  }
}

resource "google_cloud_tasks_queue" "vendor_sync_queue" {
  name     = "vendor-sync-queue"
  location = var.region
  rate_limits {
    max_dispatches_per_second = 5
    max_concurrent_dispatches = 10
  }
  retry_config {
    max_attempts = 5
    min_backoff  = "10s"
    max_backoff  = "300s"
  }
}
Secret Manager rotation
resource "google_secret_manager_secret" "vendor_api_key" {
  secret_id = "vendor-api-key"
  replication { auto {} }
  rotation {
    rotation_period    = "2592000s"  # 30日
    next_rotation_time = timeadd(timestamp(), "720h")
  }
  topics {
    name = google_pubsub_topic.secret_rotation.id
  }
}

# 対象シークレットのみ secretAccessor
resource "google_secret_manager_secret_iam_member" "job_sa_secret_access" {
  secret_id = google_secret_manager_secret.vendor_api_key.id
  role      = "roles/secretmanager.secretAccessor"
  member    = "serviceAccount:${google_service_account.vendor_sync_job_sa.email}"
}

アーキテクチャ図 — Cloud Run Jobs × Cloud Tasks × Secret Manager rotation × Scheduler OIDC

Cloud Scheduler 0 3 * * * (JST) OIDC token 付与 Google Cloud SA: vendor-sync-scheduler role: run.invoker(Job限定) OIDC で Jobs Execution API 起動 Cloud Run Jobs — vendor-sync-job(実行時間のみ課金) task_count=1 / parallelism=1 / timeout=3600s / max_retries=3 SA: vendor-sync-job(専用・最小権限) Before: Service min-instances=1 常時起動 Secret Manager secret: vendor-api-key rotation_period: 2592000s(30日) IAM: secretAccessor(対象シークレットのみ) Before: revision env var に平文 rotationは通知のみ・自動更新ではない Pub/Sub topic: secret-rotation-notify sub: secret-rotation-handler → Cloud Function(新キー発行・格納) (別モジュール・本問スコープ外) Cloud Tasks — vendor-sync-queue rate_limits max_dispatches_per_second: 5 max_concurrent_dispatches: 10 retry_config(指数バックオフ) min_backoff:10s → max_backoff:300s max_attempts:5 / max_doublings:4 ベンダー API(外部・レート上限あり) Before: forループ無制限リクエスト → 429多発 After: Cloud Tasks 経由で5 req/秒に整流 Terraform: cloud_run_v2_job / cloud_tasks_queue / secret_manager_secret / secret_manager_secret_iam_member / cloud_scheduler_job OIDC 起動 シークレット取得 30日毎に通知 SKUごとにタスクをenqueue 5 req/秒で dispatch

模範解答 B

# terraform/modules/vendor-sync/main.tf
# Cloud Run Jobs × Cloud Tasks × Secret Manager rotation × Cloud Scheduler(OIDC)

locals {
  project_id       = var.project_id
  region           = var.region
  vendor_secret_id = "vendor-api-key"
}

# ── 1. Secret Manager — ベンダー API キー + ローテーション通知 ──────────────────
# Before: Cloud Run revision の環境変数に平文設定(閲覧可能・ローテーションなし)
resource "google_pubsub_topic" "secret_rotation" {
  name    = "secret-rotation-notify"
  project = local.project_id
}

resource "google_secret_manager_secret" "vendor_api_key" {
  secret_id = local.vendor_secret_id
  project   = local.project_id

  replication {
    auto {}
  }

  # 30日ごとにローテーション通知(実際のキー更新は通知を受けたハンドラーで実施)
  rotation {
    rotation_period    = "2592000s" # 30日
    next_rotation_time = timeadd(timestamp(), "720h")
  }

  topics {
    name = google_pubsub_topic.secret_rotation.id
  }
}

# ── 2. Job 実行用サービスアカウント(最小権限) ─────────────────────────────────
# Before: デフォルト SA(プロジェクト編集者相当)
resource "google_service_account" "vendor_sync_job_sa" {
  account_id   = "vendor-sync-job"
  display_name = "Vendor Sync Cloud Run Job SA"
  project      = local.project_id
}

# 対象シークレットのみ secretAccessor(プロジェクト全体には付与しない)
resource "google_secret_manager_secret_iam_member" "job_sa_secret_access" {
  secret_id = google_secret_manager_secret.vendor_api_key.id
  role      = "roles/secretmanager.secretAccessor"
  member    = "serviceAccount:${google_service_account.vendor_sync_job_sa.email}"
}

# Cloud Tasks へのタスク投入権限のみ
resource "google_project_iam_member" "job_sa_tasks_enqueuer" {
  project = local.project_id
  role    = "roles/cloudtasks.enqueuer"
  member  = "serviceAccount:${google_service_account.vendor_sync_job_sa.email}"
}

# ── 3. Cloud Tasks キュー — レート制御 + 指数バックオフ ─────────────────────────
# Before: for ループで無制限に同時リクエスト → ベンダーAPI 429 多発
resource "google_cloud_tasks_queue" "vendor_sync_queue" {
  name     = "vendor-sync-queue"
  location = local.region
  project  = local.project_id

  rate_limits {
    max_dispatches_per_second = 5   # ベンダーAPIのレート上限に合わせる
    max_concurrent_dispatches = 10
  }

  retry_config {
    max_attempts  = 5
    min_backoff   = "10s"
    max_backoff   = "300s"
    max_doublings = 4               # 指数バックオフ: 10s→20s→40s→80s→160s
  }
}

# ── 4. Cloud Run Jobs — バッチ本体 ─────────────────────────────────────────────
# Before: Cloud Run Service (min-instances=1) で24時間常時起動課金
# After : Cloud Run Jobs は実行時間のみ課金(実行は深夜数分のみ)
resource "google_cloud_run_v2_job" "vendor_sync_job" {
  name     = "vendor-sync-job"
  location = local.region
  project  = local.project_id

  template {
    task_count  = 1
    parallelism = 1

    template {
      service_account = google_service_account.vendor_sync_job_sa.email
      max_retries      = 3
      timeout          = "3600s"

      containers {
        image = "asia-northeast1-docker.pkg.dev/${local.project_id}/mops/vendor-sync:latest"

        env {
          # Secret 名のみを渡し、値は実行時に Secret Manager API 経由で取得する
          name  = "VENDOR_API_KEY_SECRET"
          value = google_secret_manager_secret.vendor_api_key.secret_id
        }
        env {
          name  = "TASKS_QUEUE"
          value = google_cloud_tasks_queue.vendor_sync_queue.id
        }

        resources {
          limits = {
            cpu    = "1"
            memory = "512Mi"
          }
        }
      }
    }
  }
}

# ── 5. Cloud Scheduler — OIDC 認証で Jobs Execution API を起動 ─────────────────
resource "google_service_account" "scheduler_invoker_sa" {
  account_id   = "vendor-sync-scheduler"
  display_name = "Vendor Sync Scheduler Invoker"
  project      = local.project_id
}

resource "google_cloud_run_v2_job_iam_member" "scheduler_invoker" {
  name     = google_cloud_run_v2_job.vendor_sync_job.name
  location = local.region
  project  = local.project_id
  role     = "roles/run.invoker"
  member   = "serviceAccount:${google_service_account.scheduler_invoker_sa.email}"
}

resource "google_cloud_scheduler_job" "vendor_sync_nightly" {
  name     = "vendor-sync-nightly"
  project  = local.project_id
  region   = local.region
  schedule = "0 3 * * *"  # 毎日 3:00 JST

  http_target {
    http_method = "POST"
    uri         = "https://${local.region}-run.googleapis.com/apis/run.googleapis.com/v1/namespaces/${local.project_id}/jobs/${google_cloud_run_v2_job.vendor_sync_job.name}:run"

    oauth_token {
      service_account_email = google_service_account.scheduler_invoker_sa.email
    }
  }
}

Bad vs Good 設計比較

観点Bad(Cloud Run Service 常駐)Good(Cloud Run Jobs)
実行基盤 min-instances=1 で24時間課金 Cloud Run Jobs で実行時間のみ課金
APIキー管理 Revision の環境変数に平文(閲覧可能) Secret Manager + rotation_period 30日 + Pub/Sub通知
レート制御 for ループで無制限リクエスト → 429多発 Cloud Tasks max_dispatches_per_second=5
リトライ アプリ内 try/except で握りつぶし Cloud Tasks retry_config 指数バックオフ
IAM デフォルトSA(プロジェクト編集者相当) 専用SA + secretAccessor/cloudtasks.enqueuer のみ
トリガー認証 Scheduler → HTTP(認証設定漏れのリスク) Scheduler → Jobs Execution API(OIDC専用SA)

確認コマンド

# 1. Job 実行履歴の確認
gcloud run jobs executions list --job=vendor-sync-job --region=asia-northeast1
# Expected: 深夜3時台の実行が数分で Succeeded になっていること

# 2. Secret のローテーション設定確認
gcloud secrets describe vendor-api-key \
  --format="value(rotation.rotationPeriod,rotation.nextRotationTime)"
# Expected: rotationPeriod=2592000s(30日)

# 3. Cloud Tasks キューのレート設定確認
gcloud tasks queues describe vendor-sync-queue --location=asia-northeast1 \
  --format="value(rateLimits.maxDispatchesPerSecond,rateLimits.maxConcurrentDispatches)"
# Expected: 5.0 10

# 4. Job SA の IAM 権限が最小限であることを確認
gcloud projects get-iam-policy ${PROJECT_ID} \
  --flatten="bindings[].members" \
  --filter="bindings.members:vendor-sync-job@*" \
  --format="table(bindings.role)"
# Expected: roles/cloudtasks.enqueuer のみ(roles/editor 等の広い権限が無いこと)
# secretAccessor は Secret 単位の IAM のためプロジェクトレベルには出ない

# 5. Scheduler → Jobs 起動の疎通確認
gcloud scheduler jobs run vendor-sync-nightly --location=asia-northeast1
gcloud run jobs executions list --job=vendor-sync-job --region=asia-northeast1 --limit=1
# Expected: 手動トリガー直後に新しい execution が Running → Succeeded になること

# 6. 429 エラー率の確認(Cloud Tasks 導入前後比較)
gcloud logging read \
  'resource.type="cloud_run_job" AND jsonPayload.vendor_status_code=429' \
  --freshness=1d --format="value(timestamp)" | wc -l
# Expected: レート制御導入後は 0 件近くまで減少

ポイント解説

カテゴリ A

1 ファーストクラスコレクションの意義(Ch9)
CouponBasket は単なる「クーポンの入れ物」ではなく、「クーポン群に対する操作(追加・重複防止・反復)」を1つの型に閉じ込める。生の list を外部に公開すると、どこからでも append/clear されて不整合な状態を招くため、_coupons を隠蔽し操作を専用メソッド経由に限定する。
2 不変な add_coupon(Ch4)
add_couponself を変更せず新しい CouponBasket を返すことで、「あるコードパスで basket に副作用が起きて別のコードパスの前提が崩れる」というバグを構造的に防げる。カート画面のようにUndo/再計算が頻発する箇所では特に有効。
3 match 文による条件分岐の一元化(Ch8)
if/elif の連鎖では新しい ConditionType を追加するたびに全ての分岐箇所を探して修正する必要があるが、match + StrEnum なら網羅性が視覚的に把握しやすく、raise AssertionError で未知の値をフェイルラウドにできる。

カテゴリ B

4 Cloud Run Jobs と Service の使い分け
Service は「常時リクエストを受け付ける」用途、Jobs は「開始・終了があるバッチ」用途に設計されている。深夜バッチのような実行時間が短く定期的な処理を Service で動かすと min-instances 分の待機コストが無駄になる。Jobs は task_count/parallelism で並列実行数も制御できる。
5 Secret Manager の rotation は「通知」であってキー自体の自動更新ではない
rotation_period を設定すると Pub/Sub にローテーション時期の通知が飛ぶだけで、ベンダー側の新キー発行・Secret への格納は別途ハンドラー(Cloud Function 等)で実装する必要がある。これを誤解して「rotation を設定すれば自動でキーが変わる」と思い込むと運用が破綻する。
6 Cloud Tasks によるバックプレッシャー設計
max_dispatches_per_secondmax_concurrent_dispatches の組み合わせで、下流(ベンダーAPI)の処理能力を超えないようにエンキュー側で流量を絞る。retry_config の指数バックオフにより一時的な429/5xxはアプリコードで個別に try/except せずインフラ層で吸収できる。

実務への応用

  • CouponBasket のようなファーストクラスコレクションは、MOps のキャンペーン対象者フィルタやポイント付与ルールなど「ルールの集合」を扱う箇所全般に転用できる: 生の list[dict] でルールを持ち回るコードを見つけたら、専用の値オブジェクト + 隠蔽コレクションに置き換えることでバグの温床を先に潰せる
  • Cloud Run Jobs への移行は「常時起動 Service で動いているバッチもどき」を洗い出す良い機会になる: gcloud run services list で min-instances≥1 かつ Cloud Scheduler からしか叩かれていないサービスを棚卸しし、Jobs 化することで待機コストを削減できる
  • Secret Manager rotation + Pub/Sub 通知パターンは、ベンダーAPIキーだけでなく DB パスワードや DataDog API キーの定期更新運用にも同じ形で適用できる: 通知を受けるハンドラー(Cloud Function)を1つ共通化しておくと横展開が容易
  • Cloud Tasks のレート制御は、ベンダーとの契約レート上限が明文化されている外部連携すべてに適用すべき標準パターン: for ループで直接叩くコードは短期的には動くが、ベンダー側のトラフィック増加時に真っ先に障害の原因になる

今日のまとめ

カテゴリAでは、生の list[dict] で表現されたクーポンスタッキングロジックを、StrEnum + frozen dataclass の値オブジェクト群と、_coupons を隠蔽したファーストクラスコレクション CouponBasket に置き換えることで、3重ネストループとマジックストリングを解消し、add_coupon の不変設計によって状態不整合を構造的に防いだ(Ch3/Ch4/Ch8/Ch9/Ch10)。

カテゴリBでは、常時起動していた Cloud Run Service を Cloud Run Jobs に移行して待機コストを排除し、Cloud Tasks のレート制御でベンダーAPIの429エラーを構造的に防ぎ、Secret Manager rotation + Pub/Sub 通知でAPIキーの平文管理から脱却した。どちらも共通するのは「無秩序に散らばった責務を専用の型・専用のマネージドリソースに閉じ込める」という設計思想である。

自己評価

自分の回答

気づき・メモ