2026-06-27 A+B: @singledispatch × __init_subclass__ × TypeIs（注文イベントハンドラ Bad→Good）× Argo Workflows × VPC Service Controls（サービス境界 + restricted.googleapis.com + Cloud Run Jobs 最小権限 Terraform）

概要

🔀

@singledispatch で OCP（Open/Closed 原則）を実現

functools.singledispatch は第1引数の実行時型でディスパッチを行う。@handle_event.register(OrderPlacedEvent) のように型で登録し、新しいイベントを追加するときは既存コードを一切変更せず新クラスと新ハンドラを書くだけで良い。if-elif チェーンは Open/Closed 原則違反 — 追加のたびにディスパッチコードを変更する必要がある。

🪝

__init_subclass__ でサブクラスを自動登録

class OrderPlacedEvent(OrderEvent, event_type="placed"): と書いた瞬間に __init_subclass__(cls, event_type="placed") が呼ばれ、_registry["placed"] = OrderPlacedEvent が自動で登録される。マジックストリングなしで dispatch_from_dict が動作し、登録忘れが構造的に起きない。

🔍

TypeIs[T] — 双方向ナローイング（PEP 742）

def is_order_placed(e: OrderEvent) -> TypeIs[OrderPlacedEvent]: は isinstance の戻り値型を型チェッカーに正確に伝える。TypeGuard と違い TypeIs は双方向ナローイングで、else ブランチでも残りの型が絞り込まれる。mypy / pyright が event.amount を Decimal として正確に推論できる。

🛡️

VPC SC — restricted.googleapis.com が VPC SC バイパス防止の核心

private.googleapis.com（199.36.153.8/30）はアクセスを境界チェックしない。restricted.googleapis.com（199.36.153.4/30）を使うと VPC SC の境界チェックを通過するため、PII データセットへの外部アクセスを確実に 403 で拒否できる。DNS CNAME で *.googleapis.com → restricted.googleapis.com に向ける設定が必須。

問題 A: コーディング — @singledispatch × __init_subclass__ × TypeIs（注文イベントハンドラ Bad→Good）

以下の「悪いコード」は、ECサイト MOps チームの注文イベントハンドラです。問題点を全て洗い出し、@singledispatch・__init_subclass__ + ClassVar・TypeIs（PEP 742）・frozen dataclass + slots=True・StrEnum を使って Bad→Good にリファクタリングしてください。

制約・前提条件

Python 3.12+（@singledispatch は functools.singledispatch、TypeIs は typing.TypeIs）
@singledispatch で OrderEvent のサブタイプに応じたハンドラを型安全にディスパッチすること
__init_subclass__ でサブクラス登録を自動化し、マジックストリングなしで event_type を ClassVar として管理すること
TypeIs[T] で型ナローイングガードを実装し、isinstance チェックを型安全にすること
処理結果を @dataclass(frozen=True, slots=True) HandlerResult で返すこと
Google スタイル docstring・インラインコメント・名前付き定数を含めること

期待する回答形式: 問題点の列挙（番号付き）+ 改善後コード + 実行例（input→output）+ 適用した設計パターン名と書籍対応章

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

このコードには 7つの設計上の問題 が隠れています。見つけてみてください。

bad_order_handler.py — マジックストリング・isinstance 乱用・型なし

import json

class OrderEvent:
    def __init__(self, event_type, data):
        self.event_type = event_type  # 問題1: マジックストリング
        self.data = data              # 問題2: dict（型なし）

def handle_event(event):
    # 問題3: if-elif チェーン（OCP 違反）
    if event.event_type == "placed":
        return {"status": "ok",
                "msg": "Order placed: " + str(event.data.get("order_id"))}
    elif event.event_type == "shipped":
        return {"status": "ok",
                "msg": "Shipped: " + str(event.data.get("tracking_no"))}
    elif event.event_type == "cancelled":
        # 問題4: エラー処理なし（reason=None でも素通り）
        return {"status": "ok",
                "msg": "Cancelled: " + str(event.data.get("reason"))}
    else:
        # 問題5: 未知イベントを握りつぶし（サイレント失敗）
        return {"status": "unknown"}

# 問題6: 結果が dict（型安全性なし）
# 問題7: ハンドラとディスパッチが密結合（テスト不可）

問題点サマリー（7点）

1マジックストリング event_type（Ch2/Ch7） — "placed" / "shipped" が散在。タイポで実行時 KeyError になる。ClassVar event_type + __init_subclass__ で自動設定

2data が dict（型なし・スキーマ不明）（Ch2） — event.data.get("order_id") は型安全でない。frozen dataclass OrderPlacedEvent(order_id: str, amount: Decimal) に変える

3if-elif チェーンが OCP 違反（Ch6） — 新イベント追加のたびに handle_event を変更する。@singledispatch でディスパッチを型に委譲

4バリデーションなし（Ch2/Ch10） — reason=None・amount=-1 でも素通り。__post_init__ で値域チェック

5未知イベントを握りつぶし（Ch10） — return {"status": "unknown"} はサイレント失敗。@singledispatch デフォルトで ValueError を raise（Fail-Fast）

6結果が dict（型安全性なし）（Ch4） — {"status": "ok", "msg": ...} はプロパティなし・型なし。frozen dataclass HandlerResult + summary property に変える

7ハンドラとディスパッチが密結合（Ch11） — handle_event が全イベントのロジックを知りすぎている。@singledispatch + @handle_event.register で分離しテスタブルにする

ヒント A（段階的開示）

ヒント1 — 方向性

@singledispatch はファーストクラス引数の型でディスパッチを行う。OrderEvent 基底クラスのデフォルトハンドラを登録し、サブクラスごとに @handle_event.register(OrderPlacedEvent) で特化ハンドラを追加する。__init_subclass__ を使うと OrderEvent のサブクラスが定義された瞬間に cls.event_type（ClassVar）を自動で _registry に登録できる。新イベントを追加するときは if-elif を触らず新クラスを書くだけで良くなる（OCP）。TypeIs[T] は双方向ナローイングで else ブランチでも型が絞り込まれる。

ヒント2 — アプローチ

class OrderEvent に _registry: ClassVar[dict[str, type[OrderEvent]]] = {} を持たせる
__init_subclass__(cls, event_type: str, **kw) で cls.event_type = event_type を設定し _registry[event_type] = cls に登録
TypeIs[T] で def is_order_placed(e: OrderEvent) -> TypeIs[OrderPlacedEvent]: を実装
@singledispatch の handle_event(event: OrderEvent) を定義し、各サブクラスに @handle_event.register でハンドラを追加
@dataclass(frozen=True, slots=True) の HandlerResult に success: bool・message: str・event_type: str を持たせ summary property を追加
CancelReason(StrEnum) で reason フィールドを型安全に

ヒント3 — コードの骨格

@dataclass(frozen=True, slots=True)
class HandlerResult:
    success: bool
    message: str
    event_type: str

    @property
    def summary(self) -> str:
        status = "OK" if self.success else "NG"
        return f"[{status}] {self.event_type}: {self.message}"


class OrderEvent:
    _registry: ClassVar[dict[str, type["OrderEvent"]]] = {}
    event_type: ClassVar[str]

    def __init_subclass__(cls, event_type: str, **kw: object) -> None:
        super().__init_subclass__(**kw)
        cls.event_type = event_type             # ← ClassVar へ自動設定
        OrderEvent._registry[event_type] = cls   # ← 自動登録（OCP）

@dataclass(frozen=True, slots=True)
class OrderPlacedEvent(OrderEvent, event_type="placed"):
    order_id: str
    amount: Decimal

@singledispatch
def handle_event(event: OrderEvent) -> HandlerResult:
    raise ValueError(f"未知のイベント型: {type(event).__name__!r}")

@handle_event.register
def _(event: OrderPlacedEvent) -> HandlerResult:
    return HandlerResult(success=True,
                         message=f"注文確定: {event.order_id}",
                         event_type=event.event_type)

# TypeIs 型ガード
def is_order_placed(e: OrderEvent) -> TypeIs[OrderPlacedEvent]:
    return isinstance(e, OrderPlacedEvent)

問題点分析 — カテゴリ A

#	問題点	分類	改善方法
1	マジックストリング event_type	型の活用 Ch2/Ch7	`ClassVar event_type` + `__init_subclass__`
2	data が dict（型なし）	型の活用 Ch2	`frozen dataclass` サブクラスでフィールド型付け
3	if-elif チェーン（OCP 違反）	条件分岐 Ch6	`@singledispatch` でポリモーフィックディスパッチ
4	バリデーションなし	エラー処理 Ch10	`__post_init__` で値域チェック
5	未知イベントを握りつぶし	エラー処理 Ch10	デフォルトハンドラで `ValueError`（Fail-Fast）
6	結果が dict（型安全性なし）	型の活用 Ch4	`frozen dataclass HandlerResult` + `summary` property
7	ハンドラとディスパッチが密結合	テスト容易性 Ch11	`@singledispatch` + `@handle_event.register` で分離

模範解答 A

Before — マジックストリング・if-elif・dict・握りつぶし

class OrderEvent:
    def __init__(self, event_type, data):
        self.event_type = event_type  # マジックストリング
        self.data = data              # dict（型なし）

def handle_event(event):
    if event.event_type == "placed":     # OCP 違反
        return {"status": "ok",
                "msg": "Order placed: " + str(event.data.get("order_id"))}
    elif event.event_type == "shipped":
        return {"status": "ok",
                "msg": "Shipped: " + str(event.data.get("tracking_no"))}
    elif event.event_type == "cancelled":
        return {"status": "ok",          # バリデーションなし
                "msg": "Cancelled: " + str(event.data.get("reason"))}
    else:
        return {"status": "unknown"}     # サイレント失敗

After — @singledispatch × __init_subclass__ × TypeIs × frozen dataclass

"""order_handler.py — Ch2/Ch4/Ch6/Ch7/Ch10/Ch11"""
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
from enum import StrEnum
from functools import singledispatch
from typing import ClassVar, Final, TypeIs
import logging

logger = logging.getLogger(__name__)

MAX_ORDER_AMOUNT: Final[Decimal] = Decimal("9_999_999")
MIN_ORDER_ID_LEN: Final[int] = 6
MIN_TRACKING_LEN: Final[int] = 8


class CancelReason(StrEnum):
    """キャンセル理由（マジックストリング廃止）。"""
    OUT_OF_STOCK      = "out_of_stock"
    CUSTOMER_REQUEST  = "customer_request"
    PAYMENT_FAILED    = "payment_failed"


@dataclass(frozen=True, slots=True)
class HandlerResult:
    """イベントハンドリング結果の値オブジェクト。"""
    success:    bool
    message:    str
    event_type: str

    @property
    def summary(self) -> str:
        status = "OK" if self.success else "NG"
        return f"[{status}] {self.event_type}: {self.message}"


class OrderEvent:
    """注文イベント基底クラス。
    __init_subclass__ でサブクラスを _registry に自動登録する（OCP）。
    """
    _registry:  ClassVar[dict[str, type[OrderEvent]]] = {}
    event_type: ClassVar[str]

    def __init_subclass__(cls, event_type: str, **kw: object) -> None:
        super().__init_subclass__(**kw)
        cls.event_type = event_type              # ← ClassVar へ自動設定
        OrderEvent._registry[event_type] = cls   # ← 自動登録


@dataclass(frozen=True, slots=True)
class OrderPlacedEvent(OrderEvent, event_type="placed"):
    order_id: str
    amount:   Decimal

    def __post_init__(self) -> None:
        if len(self.order_id) < MIN_ORDER_ID_LEN:
            raise ValueError(f"order_id は {MIN_ORDER_ID_LEN} 文字以上: {self.order_id!r}")
        if self.amount <= 0 or self.amount > MAX_ORDER_AMOUNT:
            raise ValueError(f"amount は 0 < x <= {MAX_ORDER_AMOUNT}: {self.amount}")


@dataclass(frozen=True, slots=True)
class OrderShippedEvent(OrderEvent, event_type="shipped"):
    order_id:    str
    tracking_no: str

    def __post_init__(self) -> None:
        if len(self.tracking_no) < MIN_TRACKING_LEN:
            raise ValueError(f"tracking_no は {MIN_TRACKING_LEN} 文字以上")


@dataclass(frozen=True, slots=True)
class OrderCancelledEvent(OrderEvent, event_type="cancelled"):
    order_id: str
    reason:   CancelReason  # StrEnum で型安全（マジックストリング廃止）


# ── TypeIs 型ナローイングガード ──────────────────────────────────────────────
def is_order_placed(e: OrderEvent) -> TypeIs[OrderPlacedEvent]:
    """双方向ナローイング: else ブランチでも型が絞り込まれる（PEP 742）。"""
    return isinstance(e, OrderPlacedEvent)


# ── @singledispatch ──────────────────────────────────────────────────────────
@singledispatch
def handle_event(event: OrderEvent) -> HandlerResult:
    """未知の型: Fail-Fast で ValueError（サイレント失敗の禁止）。"""
    raise ValueError(
        f"未知のイベント型: {type(event).__name__!r}。"
        f"登録済み: {list(OrderEvent._registry.keys())}"
    )

@handle_event.register
def _(event: OrderPlacedEvent) -> HandlerResult:
    logger.info("注文確定: order_id=%s amount=%s", event.order_id, event.amount)
    return HandlerResult(
        success=True,
        message=f"注文 {event.order_id} 確定（¥{event.amount:,}）",
        event_type=event.event_type,
    )

@handle_event.register
def _(event: OrderShippedEvent) -> HandlerResult:
    logger.info("出荷完了: order_id=%s tracking=%s", event.order_id, event.tracking_no)
    return HandlerResult(
        success=True,
        message=f"注文 {event.order_id} 出荷完了（追跡番号: {event.tracking_no}）",
        event_type=event.event_type,
    )

@handle_event.register
def _(event: OrderCancelledEvent) -> HandlerResult:
    logger.warning("キャンセル: order_id=%s reason=%s", event.order_id, event.reason)
    return HandlerResult(
        success=True,
        message=f"注文 {event.order_id} キャンセル（理由: {event.reason}）",
        event_type=event.event_type,
    )


def dispatch_from_dict(raw: dict) -> HandlerResult:
    """生 dict から _registry 経由でルーティングして dispatch。"""
    event_type = raw.get("event_type", "")
    cls = OrderEvent._registry.get(event_type)
    if cls is None:
        raise KeyError(f"未登録イベント種別: {event_type!r}")
    payload = {k: v for k, v in raw.items() if k != "event_type"}
    return handle_event(cls(**payload))

from decimal import Decimal

# 正常系
placed = OrderPlacedEvent(order_id="ORD-001234", amount=Decimal("9800"))
print(handle_event(placed).summary)
# [OK] placed: 注文 ORD-001234 確定（¥9,800）

shipped = OrderShippedEvent(order_id="ORD-001234", tracking_no="JPN-TRACK-001")
print(handle_event(shipped).summary)
# [OK] shipped: 注文 ORD-001234 出荷完了（追跡番号: JPN-TRACK-001）

cancelled = OrderCancelledEvent(order_id="ORD-001234", reason=CancelReason.CUSTOMER_REQUEST)
print(handle_event(cancelled).summary)
# [OK] cancelled: 注文 ORD-001234 キャンセル（理由: customer_request）

# dict から dispatch
raw = {"event_type": "placed", "order_id": "ORD-005678", "amount": Decimal("2500")}
print(dispatch_from_dict(raw).summary)
# [OK] placed: 注文 ORD-005678 確定（¥2,500）

# 未知イベント → ValueError（Fail-Fast）
bad = object.__new__(OrderEvent)
try:
    handle_event(bad)
except ValueError as e:
    print(e)
# 未知のイベント型: 'OrderEvent'。登録済み: ['placed', 'shipped', 'cancelled']

# TypeIs 型ナローイング
event: OrderEvent = OrderPlacedEvent(order_id="ORD-999000", amount=Decimal("100"))
if is_order_placed(event):
    print(event.amount)  # mypy: Decimal（型安全）
# 100

# 値オブジェクトの不変性
result = handle_event(placed)
try:
    result.success = False  # type: ignore[misc]
except Exception:
    print("FrozenInstanceError: slots=True で変更不可")
# FrozenInstanceError: slots=True で変更不可

ポイント	適用した設計原則/パターン	書籍対応章
`@singledispatch` でポリモーフィックディスパッチ	Open/Closed 原則・多態性	Ch6
`__init_subclass__` でサブクラス自動登録	登録パターン・DRY	Ch4/Ch6
`TypeIs[T]` 双方向ナローイングガード	型の活用・型安全 isinstance	Ch2
`ClassVar event_type` でマジックストリング廃止	型の活用・名前付き定数	Ch2/Ch7
`CancelReason(StrEnum)`	マジックストリング廃止	Ch6
`@dataclass(frozen=True, slots=True) HandlerResult`	値オブジェクト・不変性	Ch4
未知イベント `ValueError`（Fail-Fast）	エラー処理	Ch10

# tests/test_order_handler.py
import pytest
from decimal import Decimal
from order_handler import (
    OrderPlacedEvent, OrderShippedEvent, OrderCancelledEvent,
    CancelReason, handle_event, dispatch_from_dict,
    HandlerResult, is_order_placed, OrderEvent, MAX_ORDER_AMOUNT,
)


class TestOrderPlacedEvent:
    def test_valid_placed(self):
        e = OrderPlacedEvent(order_id="ORD-001234", amount=Decimal("9800"))
        assert e.event_type == "placed"
        assert e.order_id == "ORD-001234"

    def test_short_order_id_raises(self):
        with pytest.raises(ValueError, match="order_id は"):
            OrderPlacedEvent(order_id="X", amount=Decimal("100"))

    def test_negative_amount_raises(self):
        with pytest.raises(ValueError, match="amount は"):
            OrderPlacedEvent(order_id="ORD-001", amount=Decimal("-1"))

    def test_over_max_amount_raises(self):
        with pytest.raises(ValueError, match="amount は"):
            OrderPlacedEvent(order_id="ORD-001234", amount=MAX_ORDER_AMOUNT + 1)


class TestHandlerResult:
    def test_summary_ok(self):
        r = HandlerResult(success=True, message="確定", event_type="placed")
        assert r.summary == "[OK] placed: 確定"

    def test_summary_ng(self):
        r = HandlerResult(success=False, message="失敗", event_type="placed")
        assert r.summary == "[NG] placed: 失敗"

    def test_frozen_immutability(self):
        r = HandlerResult(success=True, message="OK", event_type="placed")
        with pytest.raises(Exception):
            r.success = False  # type: ignore[misc]


class TestSingledispatch:
    def test_placed(self):
        e = OrderPlacedEvent(order_id="ORD-001234", amount=Decimal("5000"))
        result = handle_event(e)
        assert result.success is True
        assert "ORD-001234" in result.message

    def test_shipped(self):
        e = OrderShippedEvent(order_id="ORD-001234", tracking_no="JPN-TRACK-001")
        result = handle_event(e)
        assert "JPN-TRACK-001" in result.message

    def test_cancelled(self):
        e = OrderCancelledEvent(order_id="ORD-001234", reason=CancelReason.OUT_OF_STOCK)
        result = handle_event(e)
        assert "out_of_stock" in result.message

    def test_unknown_raises(self):
        bad = object.__new__(OrderEvent)
        with pytest.raises(ValueError, match="未知のイベント型"):
            handle_event(bad)


class TestTypeIs:
    def test_is_order_placed_true(self):
        e: OrderEvent = OrderPlacedEvent(order_id="ORD-001234", amount=Decimal("100"))
        assert is_order_placed(e) is True

    def test_is_order_placed_false(self):
        e: OrderEvent = OrderShippedEvent(order_id="ORD-001234", tracking_no="JPN-TRACK-001")
        assert is_order_placed(e) is False


class TestDispatchFromDict:
    def test_dispatch_placed(self):
        raw = {"event_type": "placed", "order_id": "ORD-001234",
               "amount": Decimal("2500")}
        result = dispatch_from_dict(raw)
        assert result.event_type == "placed"

    def test_unknown_event_type_raises(self):
        raw = {"event_type": "refunded", "order_id": "ORD-001234"}
        with pytest.raises(KeyError, match="未登録イベント種別"):
            dispatch_from_dict(raw)

問題 B: インフラ — Argo Workflows × VPC Service Controls（サービス境界 + restricted.googleapis.com）× Cloud Run Jobs 最小権限（Terraform）

ECサイト MOps チームの Argo Workflows バッチが以下の IAM・ネットワーク課題を抱えています。

現状の課題:

BigQuery・Secret Manager へのアクセスが VPC 外（パブリックエンドポイント）経由 で行われており、データ漏洩リスクがある
Argo Workflows Pod が BigQuery に対して roles/bigquery.admin（過剰権限） で接続している
機密 BigQuery データセット（mops_pii）が VPC Service Controls 境界外にある（外部からのアクセスを境界で防げない）
Cloud Run Jobs のバッチジョブが allUsers（認証なし） で起動できる設定になっている

要件

#	要件
1	VPC Service Controls でサービス境界（`mops-perimeter`）を Terraform で作成し、`bigquery.googleapis.com` と `secretmanager.googleapis.com` を境界内に含めること
2	GKE サブネットで `private_ip_google_access = true` を有効化し、restricted.googleapis.com（199.36.153.4/30）へのルートと DNS CNAME を設定して VPC SC バイパスを防ぐこと
3	Cloud Run Jobs の IAM を `allUsers → roles/run.invoker`（Argo GSA のみ）に修正すること
4	Argo Workflows の GSA を `roles/bigquery.admin → roles/bigquery.dataEditor + roles/bigquery.jobUser`（最小権限）に変更すること

ヒント B（段階的開示）

ヒント1 — 方向性

VPC Service Controls の「サービス境界」は google_access_context_manager_service_perimeter で定義し、google_access_context_manager_access_policy をプロジェクト親に紐付ける。境界内リソース（BigQuery・Secret Manager）への外部アクセスは 403 VPC_SERVICE_CONTROLS で拒否される。重要: VPC SC 環境では private.googleapis.com（199.36.153.8/30）ではなく restricted.googleapis.com（199.36.153.4/30）を使わないと VPC SC の境界チェックをバイパスしてしまう。DNS CNAME で *.googleapis.com → restricted.googleapis.com に向ける Private DNS ゾーンが必須。

ヒント2 — Terraform リソース構成

google_access_context_manager_access_policy → 組織レベルのアクセスポリシー（scopes = ["projects/{number}"] で影響範囲を絞る）
google_access_context_manager_service_perimeter → restricted_services + resources + vpc_accessible_services
google_compute_subnetwork → private_ip_google_access = true
google_compute_route → dest_range = "199.36.153.4/30"（restricted IP 範囲）
google_dns_managed_zone + google_dns_record_set → *.googleapis.com CNAME → restricted.googleapis.com
google_cloud_run_v2_job_iam_member → roles/run.invoker を invoker SA のみに付与
google_service_account_iam_member → Argo GSA が invoker SA を impersonate

ヒント3 — VPC SC + restricted.googleapis.com の骨格

VPC SC サービス境界

resource "google_access_context_manager_access_policy" "mops_policy" {
  parent = "organizations/${var.org_id}"
  title  = "mops-access-policy"
  scopes = ["projects/${var.project_number}"]
}

resource "google_access_context_manager_service_perimeter" "mops_perimeter" {
  parent = "accessPolicies/${...}"
  name   = "accessPolicies/${...}/servicePerimeters/mops_perimeter"
  title  = "mops-perimeter"
  status {
    restricted_services = [
      "bigquery.googleapis.com",
      "secretmanager.googleapis.com",
    ]
    resources = ["projects/${var.project_number}"]
  }
}

restricted.googleapis.com DNS 設定（VPC SC バイパス防止）

# 重要: private.googleapis.com ではなく restricted.googleapis.com を使う
resource "google_compute_route" "restricted_googleapis" {
  name      = "restricted-googleapis"
  network   = google_compute_network.mops_vpc.id
  dest_range = "199.36.153.4/30"  # restricted の IP 範囲
  next_hop_gateway = "default-internet-gateway"
}

resource "google_dns_record_set" "googleapis_cname" {
  name = "*.googleapis.com."
  type = "CNAME"
  ttl  = 300
  rrdatas = ["restricted.googleapis.com."]
}

アーキテクチャ図 — VPC Service Controls × restricted.googleapis.com × Cloud Run Jobs

模範解答 B

# terraform/modules/mops-vpc-sc/main.tf
# Argo Workflows × VPC Service Controls × restricted.googleapis.com × Cloud Run Jobs

locals {
  project_id     = var.project_id
  project_number = var.project_number
  org_id         = var.org_id
  region         = var.region
}

# ── 1. VPC Service Controls — アクセスポリシー（組織レベル）────────────────
resource "google_access_context_manager_access_policy" "mops_policy" {
  parent = "organizations/${local.org_id}"
  title  = "mops-access-policy"
  # scopes でプロジェクトに限定（組織全体への影響を防ぐ）
  scopes = ["projects/${local.project_number}"]
}

# ── 2. サービス境界 — bigquery + secretmanager を境界内に配置 ─────────────
resource "google_access_context_manager_service_perimeter" "mops_perimeter" {
  parent = "accessPolicies/${google_access_context_manager_access_policy.mops_policy.name}"
  name   = "accessPolicies/${google_access_context_manager_access_policy.mops_policy.name}/servicePerimeters/mops_perimeter"
  title  = "mops-perimeter"

  # 初回は dry-run で影響確認を推奨（false にすると即時有効）
  use_explicit_dry_run_spec = false

  status {
    # 境界で保護するサービス（境界外からのアクセスは 403）
    restricted_services = [
      "bigquery.googleapis.com",
      "secretmanager.googleapis.com",
    ]
    # 境界内リソース（プロジェクト番号で指定）
    resources = ["projects/${local.project_number}"]

    # VPC Accessible Services: VPC 内からしかアクセスできないサービスを明示
    vpc_accessible_services {
      enable_restriction = true
      allowed_services   = [
        "bigquery.googleapis.com",
        "secretmanager.googleapis.com",
        "container.googleapis.com",
        "cloudrun.googleapis.com",
      ]
    }
  }

  depends_on = [google_access_context_manager_access_policy.mops_policy]
}

# ── 3. VPC ネットワーク — Private Google Access + restricted.googleapis.com ─
resource "google_compute_subnetwork" "gke_subnet" {
  name          = "mops-gke-subnet"
  region        = local.region
  network       = google_compute_network.mops_vpc.id
  ip_cidr_range = "10.0.0.0/20"

  # パブリック IP なしで Google API に到達（Private Google Access）
  private_ip_google_access = true

  secondary_ip_range {
    range_name    = "gke-pods"
    ip_cidr_range = "10.4.0.0/14"
  }
  secondary_ip_range {
    range_name    = "gke-services"
    ip_cidr_range = "10.8.0.0/20"
  }
}

# 重要: VPC SC 環境では restricted.googleapis.com（199.36.153.4/30）を使う
# private.googleapis.com（199.36.153.8/30）は VPC SC 境界チェックをバイパスするため使用禁止
resource "google_compute_route" "restricted_googleapis" {
  name             = "restricted-googleapis"
  network          = google_compute_network.mops_vpc.id
  dest_range       = "199.36.153.4/30"  # restricted.googleapis.com の IP 範囲
  next_hop_gateway = "default-internet-gateway"
  priority         = 1000
  description      = "VPC SC 環境で BQ・SecretManager を restricted 経由でアクセス"
}

# Private DNS: *.googleapis.com を restricted.googleapis.com に向ける（VPC SC 必須）
resource "google_dns_managed_zone" "googleapis_private" {
  name       = "googleapis-private"
  dns_name   = "googleapis.com."
  visibility = "private"
  private_visibility_config {
    networks {
      network_url = google_compute_network.mops_vpc.id
    }
  }
}

resource "google_dns_record_set" "googleapis_a" {
  name         = "restricted.googleapis.com."
  managed_zone = google_dns_managed_zone.googleapis_private.name
  type         = "A"
  ttl          = 300
  rrdatas      = ["199.36.153.4", "199.36.153.5", "199.36.153.6", "199.36.153.7"]
}

# *.googleapis.com → restricted.googleapis.com（全 API を restricted 経由に）
resource "google_dns_record_set" "googleapis_cname" {
  name         = "*.googleapis.com."
  managed_zone = google_dns_managed_zone.googleapis_private.name
  type         = "CNAME"
  ttl          = 300
  rrdatas      = ["restricted.googleapis.com."]
}

# ── 4. Argo Workflows 用 GSA — 最小権限（admin 廃止）────────────────────────
resource "google_service_account" "argo_batch_gsa" {
  account_id   = "mops-argo-batch-sa"
  display_name = "MOps Argo Workflows Batch SA"
  description  = "Argo Workflows バッチ用 GSA（BQ 最小権限・VPC SC 境界内）"
  project      = local.project_id
}

# Before: roles/bigquery.admin → After: dataEditor + jobUser（最小権限）
resource "google_project_iam_member" "argo_bq_data_editor" {
  project = local.project_id
  role    = "roles/bigquery.dataEditor"  # テーブル読み書き
  member  = "serviceAccount:${google_service_account.argo_batch_gsa.email}"
}

resource "google_project_iam_member" "argo_bq_job_user" {
  project = local.project_id
  role    = "roles/bigquery.jobUser"  # クエリ実行（dataEditor とセット）
  member  = "serviceAccount:${google_service_account.argo_batch_gsa.email}"
}

resource "google_project_iam_member" "argo_secret_accessor" {
  project = local.project_id
  role    = "roles/secretmanager.secretAccessor"  # Secret 読み取りのみ
  member  = "serviceAccount:${google_service_account.argo_batch_gsa.email}"
}

# ── 5. Cloud Run Jobs — 認証必須（allUsers 廃止）──────────────────────────
resource "google_service_account" "mops_job_invoker_gsa" {
  account_id   = "mops-job-invoker-sa"
  display_name = "MOps Cloud Run Jobs Invoker SA"
  description  = "Argo Workflows が Cloud Run Jobs を invoke する専用 SA"
  project      = local.project_id
}

resource "google_cloud_run_v2_job" "mops_batch_job" {
  name     = "mops-batch-job"
  location = local.region
  project  = local.project_id

  template {
    template {
      service_account = google_service_account.argo_batch_gsa.email  # 最小権限 SA

      containers {
        image = "gcr.io/${local.project_id}/mops-batch:latest"
        resources { limits = { cpu = "2", memory = "2Gi" } }
      }

      # VPC アクセス経由で restricted.googleapis.com に到達
      vpc_access {
        connector = google_vpc_access_connector.mops_connector.id
        egress    = "PRIVATE_RANGES_ONLY"
      }
    }
  }
}

# Cloud Run Jobs invoke 権限: Argo GSA のみ（allUsers 廃止）
# Before: allUsers → 誰でも起動可能
# After: invoker SA のみ → Argo GSA 経由でのみ起動可能
resource "google_cloud_run_v2_job_iam_member" "mops_job_invoker" {
  project  = local.project_id
  location = local.region
  name     = google_cloud_run_v2_job.mops_batch_job.name
  role     = "roles/run.invoker"
  member   = "serviceAccount:${google_service_account.mops_job_invoker_gsa.email}"
}

# Argo Workflows GSA → invoker SA を impersonate（最小権限の連鎖）
resource "google_service_account_iam_member" "argo_impersonate_invoker" {
  service_account_id = google_service_account.mops_job_invoker_gsa.name
  role               = "roles/iam.serviceAccountTokenCreator"
  member             = "serviceAccount:${google_service_account.argo_batch_gsa.email}"
}

# VPC Access Connector（Cloud Run Jobs → VPC 通信用）
resource "google_vpc_access_connector" "mops_connector" {
  name          = "mops-vpc-connector"
  region        = local.region
  project       = local.project_id
  network       = google_compute_network.mops_vpc.id
  ip_cidr_range = "10.9.0.0/28"
  min_instances = 2
  max_instances = 10
}

Bad vs Good 設計比較

観点	Bad（現状）	Good（改善後）
BigQuery API アクセス経路	パブリックエンドポイント経由（VPC 外）→ データ漏洩リスク	restricted.googleapis.com（199.36.153.4/30）経由 → VPC SC 境界チェックを通過
Private Google Access の API エンドポイント	private.googleapis.com（199.36.153.8/30）→ VPC SC をバイパス	restricted.googleapis.com（199.36.153.4/30）→ VPC SC 境界チェックが機能
PII データセットの保護	VPC SC 境界なし → 外部からの SELECT が可能	VPC SC 境界内 → 境界外からは 403 VPC_SERVICE_CONTROLS
Argo Workflows の BQ 権限	roles/bigquery.admin（データセット削除・IAM 変更可）	roles/bigquery.dataEditor + jobUser（テーブル読み書き + クエリ実行のみ）
Cloud Run Jobs の認証	allUsers = roles/run.invoker → 誰でも起動可能	invoker SA のみ → Argo GSA 経由でのみ起動可能（認証必須）
Terraform 管理	VPC SC・DNS・IAM が手動設定 → drift 検知不可	全リソースを Terraform 管理 → plan/apply でレビュー可能

確認コマンド（gcloud + kubectl）

# 1. VPC SC サービス境界の確認
gcloud access-context-manager perimeters describe mops_perimeter \
  --policy=POLICY_NAME \
  --format="json" | jq '.status.restrictedServices'
# ["bigquery.googleapis.com", "secretmanager.googleapis.com"] が確認できれば OK

# 2. VPC SC のドライラン確認（本番適用前）
gcloud access-context-manager perimeters dry-run describe mops_perimeter \
  --policy=POLICY_NAME
# dry-run status と spec の差分を確認する

# 3. DNS 設定確認（VPC 内 Pod から）
kubectl run -it dns-check --image=busybox:latest --rm --restart=Never -- \
  nslookup bigquery.googleapis.com
# Expected: bigquery.googleapis.com は 199.36.153.4 を返す（restricted.googleapis.com 経由）
# NG: 216.58.*.* を返す場合は DNS が private 設定を見ていない

# 4. GKE Pod 内から restricted 経由で BQ にアクセスできるか確認
kubectl run -it bq-check --image=google/cloud-sdk:slim \
  --serviceaccount=mops-argo-batch-ksa -n mops --rm \
  -- bash -c "bq ls --project_id=${PROJECT_ID}"
# roles/bigquery.dataEditor があれば dataset 一覧が表示される

# 5. Cloud Run Jobs の IAM 確認（allUsers がないことを確認）
gcloud run jobs get-iam-policy mops-batch-job \
  --region=asia-northeast1 \
  --format="json" | jq '.bindings[] | select(.role == "roles/run.invoker")'
# Expected: member が serviceAccount:mops-job-invoker-sa@... のみ
# NG: allUsers が含まれていたら即削除が必要

# 6. VPC SC 違反ログの確認（境界外からのアクセス試行）
gcloud logging read \
  'protoPayload.methodName="google.cloud.bigquery.v2.JobService.InsertJob" AND protoPayload.status.code=403' \
  --project=${PROJECT_ID} \
  --format="json" | jq '.[] | {time: .timestamp, error: .protoPayload.status.message}'
# VPC_SERVICE_CONTROLS エラーが記録されていれば境界が機能している

# 7. BigQuery Admin 権限が削除されているか確認
gcloud projects get-iam-policy ${PROJECT_ID} \
  --filter="bindings.role=roles/bigquery.admin" \
  --format="json" | jq '.bindings[]'
# mops-argo-batch-sa が含まれていないことを確認

ポイント解説

カテゴリ A

1 @singledispatch で OCP を実現（Ch6）
functools.singledispatch は第1引数の実行時型でディスパッチを行う。@handle_event.register(OrderPlacedEvent) のように型で登録し、デフォルト実装は @singledispatch デコレータ直下に書く。新イベントを追加するときは handle_event.register を持つ新クラスを書くだけで既存コードを一切変更しない（OCP）。if-elif チェーンは「変更のための修正」が必要だが singledispatch は「拡張のための追記」だけで良い。

2 __init_subclass__ でサブクラス自動登録（Ch4/Ch6）
class OrderPlacedEvent(OrderEvent, event_type="placed"): と書いた瞬間に Python が OrderEvent.__init_subclass__(cls, event_type="placed") を呼び出す。cls.event_type = event_type（ClassVar への自動設定）と OrderEvent._registry[event_type] = cls（辞書への自動登録）の2行で「マジックストリングなし・登録忘れなし」を実現。dispatch_from_dict は _registry.get(event_type) だけで正しいクラスを取得できる。

3 TypeIs[T] vs TypeGuard[T]（Ch2）
Python 3.13+ / typing_extensions の TypeIs[T] は PEP 742 で導入された双方向ナローイング。TypeGuard[T] は if ブランチのみナローイング（else では元の型のまま）だが、TypeIs[T] は else ブランチでも残りの型が絞り込まれる。if is_order_placed(event): の else ブランチで mypy が「OrderPlacedEvent ではない OrderEvent」と推論できる。実運用では typing_extensions.TypeIs で 3.12 にバックポートして使う。

カテゴリ B

4 restricted.googleapis.com vs private.googleapis.com — VPC SC の最大の落とし穴
private.googleapis.com（199.36.153.8/30）は「Private Google Access」で使うエンドポイントだが VPC SC の境界チェックを行わない。PII データセットを VPC SC で保護していても private.googleapis.com 経由なら境界チェックなしでアクセスできてしまう。restricted.googleapis.com（199.36.153.4/30）を使うと全 API 呼び出しが VPC SC 境界チェックを通過する。DNS CNAME で *.googleapis.com → restricted.googleapis.com に向けることで設定漏れを防ぐ。

5 VPC SC の dry-run で影響確認を必ず行う
use_explicit_dry_run_spec = true + spec ブロックを使うと境界を「ドライランモード」で有効化できる。実際のアクセスは拒否せず Audit Logs にのみ記録する。gcloud access-context-manager perimeters dry-run describe で違反ログを確認してから use_explicit_dry_run_spec = false に切り替えて本番有効化する。いきなり本番有効化すると既存バッチが 403 になりオンコールが必要になる。

6 Cloud Run Jobs の invoker 権限設計
Cloud Run Jobs は Cloud Run Services と異なり ingress 設定がない（リクエスト受付の概念がない）。認証制御は IAM の roles/run.invoker のみで行う。allUsers に付与すると API キーなしで anyone が gcloud run jobs execute を呼べる。Argo GSA から invoker SA への serviceAccountTokenCreator（impersonate）の連鎖により、最小権限の原則を維持しながら Argo Workflows がジョブを起動できる。

実務への応用

@singledispatch は MOps の「注文→配信→分析」パイプラインで威力を発揮する: 注文確定・出荷・返品・キャンセルなど10種類以上のイベントを if-elif で分岐すると可読性が崩壊する。@singledispatch で各イベントを独立したファイルに分けると「新しいキャンペーントリガーを追加するエンジニア」が他のハンドラを一切知らなくて良くなる
__init_subclass__ + _registry は DRY な設計パターンとして応用範囲が広い: Django の ModelBase・SQLAlchemy の DeclarativeMeta・Pydantic の ModelMetaclass は全てこのパターンの応用。自前 ORM や Plugin 機構を作るときも同じ手法が使える
VPC SC は mops_pii（会員 PII データ）の必須防衛ライン: 個人情報保護法・GDPR 対応では「不正アクセスの技術的防止措置」が必要。VPC SC の境界外からの BQ SELECT を 403 で拒否することで、開発者が誤って本番 PII データを外部から取得することを構造的に防ぐ
restricted.googleapis.com の DNS 設定は Terraform で必ず管理する: 手動で設定すると新しい GKE クラスタや Cloud Run Jobs 追加時に設定漏れが発生し「なぜか BQ にアクセスできない」バグになる。Terraform の google_dns_record_set で VPC にアタッチされた Private DNS を宣言的に管理する
証券マン視点 — VPC SC の ROI: PII データ漏洩の平均制裁金（GDPR: 売上の4%または2,000万ユーロの高い方）に対して VPC SC 設定工数は 1〜2 人日（10〜20 万円）。年商 10 億円の EC サイトでは最大 4,000 万円のリスク低減を 0.03〜0.05% のコストで実現できる

今日のまとめ

@singledispatch + __init_subclass__ で注文イベントのポリモーフィックディスパッチを実現し（if-elif チェーン → OCP 準拠）、TypeIs[T]（PEP 742）による双方向型ナローイングガードで mypy/pyright が正確に型を推論できる Python 3.12 イディオムを組み合わせることで、型安全・テスタブル・拡張容易な注文ハンドラを実装できる（Ch2/Ch4/Ch6/Ch7/Ch10/Ch11）。

インフラ側では VPC Service Controls の restricted.googleapis.com（199.36.153.4/30）+ DNS CNAME 設定が「VPC SC バイパス防止」の核心であり（private.googleapis.com を使うと VPC SC をバイパスしてしまう）、Argo Workflows の roles/bigquery.admin → dataEditor + jobUser 最小権限化と Cloud Run Jobs の allUsers → invoker SA のみ 権限絞り込みを Terraform で管理することで、MOps PII バッチ基盤のゼロトラストセキュリティモデルを宣言的に構築できる。

自己評価

完全理解おおむね理解要復習

A+B 複合 — @singledispatch × __init_subclass__ × TypeIs（Ch2/Ch4/Ch6/Ch7/Ch10/Ch11）× Argo Workflows × VPC Service Controls × restricted.googleapis.com × Cloud Run Jobs 最小権限（Terraform）

概要

@singledispatch で OCP（Open/Closed 原則）を実現

__init_subclass__ でサブクラスを自動登録

TypeIs[T] — 双方向ナローイング（PEP 742）

VPC SC — restricted.googleapis.com が VPC SC バイパス防止の核心

問題 A: コーディング — @singledispatch × __init_subclass__ × TypeIs（注文イベントハンドラ Bad→Good）

制約・前提条件

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

ヒント A（段階的開示）

問題点分析 — カテゴリ A

模範解答 A

問題 B: インフラ — Argo Workflows × VPC Service Controls（サービス境界 + restricted.googleapis.com）× Cloud Run Jobs 最小権限（Terraform）

要件

ヒント B（段階的開示）

アーキテクチャ図 — VPC Service Controls × restricted.googleapis.com × Cloud Run Jobs

模範解答 B

Bad vs Good 設計比較

確認コマンド（gcloud + kubectl）

ポイント解説

カテゴリ A

カテゴリ B

実務への応用

今日のまとめ

自己評価

自分の回答

気づき・メモ