概要
ファーストクラスコレクションで「ルールの集合」を隠蔽する
生の list を公開すると外部から自由に append/clear され不整合な状態を作れる。CouponBasket のように内部の tuple を隠蔽し、操作を専用メソッド(add_coupon)経由に限定することで、状態変化の経路を1箇所に集約できる。
不変な add_coupon で状態不整合を構造的に防ぐ
add_coupon が self を変更せず新しい 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+(
StrEnumはenum.StrEnum、match文は Python 3.10+) CouponType/ConditionTypeをStrEnumで定義し、マジックストリングを排除することCoupon/Condition/Cart/CartItemを@dataclass(frozen=True, slots=True)の値オブジェクトとして実装することCouponBasketはファーストクラスコレクション(_coupons: tuple[Coupon, ...]を内部に隠蔽し、外部に生のlistを公開しない)とし、add_couponは新しいCouponBasketを返す不変設計にすることCondition.is_satisfiedはmatch文で条件種別ごとに判定し、ネストループを解消すること- 割引合計は
cart.totalを超えないようクランプすること(マイナス残高防止) - Google スタイル docstring・インラインコメント・名前付き定数を含めること
悪いコード (Before) — カテゴリ A
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
basket.coupons.append(...) のように直接 mutate でき不整合な状態を作れる。ファーストクラスコレクションとして隠蔽するcoupon["rate"] のようなキーアクセスは typo しても実行時まで検出されない。frozen dataclass の値オブジェクトにする"percent", "fixed", "min_amount", "category" がリテラルとして各所に書かれているヒント 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 Conditionにis_satisfied(self, cart: Cart) -> boolをmatch 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_discountはmin(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
| # | 問題点 | 分類 | 改善方法 |
|---|---|---|---|
| 1 | CouponBasket.coupons が public list | カプセル化 Ch9 | ファーストクラスコレクションで隠蔽 |
| 2 | Coupon/Condition が dict 表現 | 型の活用 Ch2 | frozen dataclass の値オブジェクト化 |
| 3 | 3重ネストループ | コレクション Ch9 | 条件判定の委譲 + all() で平坦化 |
| 4 | 割引合計の上限チェックなし | 設計の悪魔 Ch10 | min(total_discount, cart.total) でクランプ |
| 5 | 重複適用防止が緩い | 設計の悪魔 Ch10 | add_coupon 側でID重複を検出 |
| 6 | マジックストリングの散在 | 設計の悪魔 Ch10 | StrEnum で定義し尽くす |
| 7 | add_coupon にバリデーションなし | 完全コンストラクタ Ch3 | 不変な add_coupon + 型検証 |
模範解答 A
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 # 上限チェックなし
"""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 Service(
min-instances=1で常時起動)を Cloud Scheduler の HTTP トリガーで叩いている → 実行時間は深夜の数分間だけなのに24時間課金が発生している - ベンダー API キーが Cloud Run revision の環境変数に平文で設定されている(
gcloud run services describeで誰でも閲覧可能)→ ローテーションの仕組みもない - ハンドラー内で約5,000 SKU 分を for ループで同時に無制限リクエスト → ベンダー API のレート制限に頻繁に抵触し429エラーで同期が不完全に終わる
- 実行用サービスアカウントがデフォルトSA(プロジェクト編集者相当の過剰権限)のまま
要件
| # | 要件 |
|---|---|
| 1 | google_cloud_run_v2_job(Cloud Run Jobs)に移行し、実行時間のみ課金される構成にすること |
| 2 | google_cloud_tasks_queue でベンダー API 呼び出しをキューイングし、rate_limits でレート制御、retry_config で指数バックオフを設定すること |
| 3 | ベンダー API キーを google_secret_manager_secret に格納し、rotation + Pub/Sub 通知でローテーション運用を組み込むこと。Job は専用SAで secretAccessor のみ付与すること |
| 4 | Cloud Scheduler は OIDC 認証の専用サービスアカウントで Cloud Run Jobs Execution API を起動すること(IAM 最小権限) |
ヒント B(段階的開示)
ヒント1 — 方向性
min-instances の概念がなく実行時間のみ課金される。Secret Manager の rotation はキー自体を自動生成しない(あくまで「そろそろ更新してください」という Pub/Sub 通知機能)ため、通知を受けて新キーを取得・格納する運用フロー(別 Cloud Function 等)とセットで設計する。Cloud Tasks は「エンキュー→ワーカーが一定レートで処理」というバックプレッシャーの標準パターン。
ヒント2 — Terraform リソース構成
google_cloud_run_v2_job(template.template.containersの入れ子構造に注意)google_cloud_tasks_queue(rate_limits+retry_config)google_secret_manager_secret(rotation.rotation_period+topics)+google_secret_manager_secret_iam_membergoogle_cloud_scheduler_job(http_target.oauth_token.service_account_emailで OIDC 起動)- Job用SAとScheduler用SAを分離し、それぞれ最小権限のロールのみ付与する
ヒント3 — リソースの骨格
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"
}
}
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
模範解答 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
CouponBasket は単なる「クーポンの入れ物」ではなく、「クーポン群に対する操作(追加・重複防止・反復)」を1つの型に閉じ込める。生の list を外部に公開すると、どこからでも append/clear されて不整合な状態を招くため、_coupons を隠蔽し操作を専用メソッド経由に限定する。
add_coupon が self を変更せず新しい CouponBasket を返すことで、「あるコードパスで basket に副作用が起きて別のコードパスの前提が崩れる」というバグを構造的に防げる。カート画面のようにUndo/再計算が頻発する箇所では特に有効。
if/elif の連鎖では新しい ConditionType を追加するたびに全ての分岐箇所を探して修正する必要があるが、match + StrEnum なら網羅性が視覚的に把握しやすく、raise AssertionError で未知の値をフェイルラウドにできる。
カテゴリ B
Service は「常時リクエストを受け付ける」用途、Jobs は「開始・終了があるバッチ」用途に設計されている。深夜バッチのような実行時間が短く定期的な処理を Service で動かすと
min-instances 分の待機コストが無駄になる。Jobs は task_count/parallelism で並列実行数も制御できる。
rotation_period を設定すると Pub/Sub にローテーション時期の通知が飛ぶだけで、ベンダー側の新キー発行・Secret への格納は別途ハンドラー(Cloud Function 等)で実装する必要がある。これを誤解して「rotation を設定すれば自動でキーが変わる」と思い込むと運用が破綻する。
max_dispatches_per_second と max_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 ループで直接叩くコードは短期的には動くが、ベンダー側のトラフィック増加時に真っ先に障害の原因になる
今日のまとめ
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キーの平文管理から脱却した。どちらも共通するのは「無秩序に散らばった責務を専用の型・専用のマネージドリソースに閉じ込める」という設計思想である。