A+B 複合 — 状態機械 × 遷移テーブル（Ch7）× GKE Workload Identity + External Secrets Operator

📋 コピー"""order_state.py — 注文ステータス遷移管理（状態機械パターン）。

良いコード・悪いコードで学ぶ設計入門（改訂新版）
  - Ch2: 型の活用（StrEnum, 型ヒント, frozen dataclass）
  - Ch7: コレクション操作の隠蔽（内部コレクションを直接公開しない）
  - Ch5: 条件分岐の削減（遷移テーブルで if-elif を廃止）
"""
from __future__ import annotations

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


# ── 定数 ─────────────────────────────────────────────────────────────────
INITIAL_STATUS: Final[str] = "pending"  # マジックストリング排除


# ── 列挙型（Ch2: 型の活用）────────────────────────────────────────────────
class OrderStatus(StrEnum):
    """注文ステータス。文字列互換でありつつ型安全。"""
    PENDING   = "pending"
    CONFIRMED = "confirmed"
    SHIPPED   = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"


# ── 遷移テーブル（Ch5: ルール = データ）─────────────────────────────────────
# 新ステータス追加時はここに1行追加するだけ（OCP）
VALID_TRANSITIONS: Final[dict[OrderStatus, frozenset[OrderStatus]]] = {
    OrderStatus.PENDING:   frozenset({OrderStatus.CONFIRMED, OrderStatus.CANCELLED}),
    OrderStatus.CONFIRMED: frozenset({OrderStatus.SHIPPED,   OrderStatus.CANCELLED}),
    OrderStatus.SHIPPED:   frozenset({OrderStatus.DELIVERED}),
    OrderStatus.DELIVERED: frozenset(),   # 終端状態: frozenset() = 遷移先なし
    OrderStatus.CANCELLED: frozenset(),   # 終端状態
}


# ── カスタム例外 ─────────────────────────────────────────────────────────
class OrderNotFoundError(KeyError):
    """指定された注文IDが存在しない場合の例外。"""


class InvalidTransitionError(ValueError):
    """許可されていないステータス遷移が要求された場合の例外。"""


# ── 値オブジェクト（Ch3 + Ch4）─────────────────────────────────────────────
@dataclass(slots=True, frozen=True)  # frozen=True で不変、slots=True でメモリ効率化
class Order:
    """注文を表す値オブジェクト。不変（frozen=True）で副作用なし。

    Attributes:
        order_id: 注文ID。
        status: 現在のステータス。
    """
    order_id: str
    status: OrderStatus


# ── 状態管理クラス（Ch7: コレクション操作の隠蔽）────────────────────────────
@dataclass
class OrderStateManager:
    """注文ステータスの状態機械。

    内部コレクション（_orders）を直接公開せず、
    意図が明確なメソッド経由でのみ操作する（Ch7）。
    """
    _orders: dict[str, OrderStatus] = field(default_factory=dict)

    def add_order(self, order_id: str) -> None:
        """新規注文を初期ステータス（PENDING）で登録する。

        Args:
            order_id: 注文ID。

        Raises:
            ValueError: 既に存在する注文IDが指定された場合。
        """
        if order_id in self._orders:
            raise ValueError(f"Order already exists: {order_id!r}")
        self._orders[order_id] = OrderStatus.PENDING  # 定数経由（マジックストリング排除）

    def transition(self, order_id: str, new_status: OrderStatus) -> None:
        """注文ステータスを遷移させる。

        Args:
            order_id: 注文ID。
            new_status: 遷移先ステータス。

        Raises:
            OrderNotFoundError: 注文IDが存在しない場合。
            InvalidTransitionError: 許可されていない遷移が要求された場合。

        Example:
            >>> mgr = OrderStateManager()
            >>> mgr.add_order("ORD-001")
            >>> mgr.transition("ORD-001", OrderStatus.CONFIRMED)
        """
        current = self._get_or_raise(order_id)
        allowed = VALID_TRANSITIONS[current]  # テーブルを引くだけ（if-elif なし）

        if new_status not in allowed:
            raise InvalidTransitionError(
                f"Cannot transition order {order_id!r}: "
                f"{current.value!r} -> {new_status.value!r}. "
                f"Allowed next: {[s.value for s in allowed] or 'none (terminal state)'}"
            )
        self._orders[order_id] = new_status

    def get_status(self, order_id: str) -> OrderStatus:
        """指定注文の現在ステータスを返す。

        Args:
            order_id: 注文ID。

        Returns:
            現在のステータス。

        Raises:
            OrderNotFoundError: 注文IDが存在しない場合。
        """
        return self._get_or_raise(order_id)

    def get_orders_by_status(self, status: OrderStatus) -> list[str]:
        """指定ステータスの注文ID一覧を返す（Ch7: コレクション操作の隠蔽）。

        Args:
            status: フィルタリングするステータス。

        Returns:
            該当する注文IDのリスト。
        """
        # リスト内包表記で外部にコレクション操作を漏らさない（Ch7）
        return [oid for oid, s in self._orders.items() if s == status]

    def snapshot(self) -> list[Order]:
        """全注文の不変スナップショットを返す。

        Returns:
            Order 値オブジェクトのリスト（コピー）。
        """
        return [Order(oid, s) for oid, s in self._orders.items()]

    def _get_or_raise(self, order_id: str) -> OrderStatus:
        """order_id が存在しない場合に OrderNotFoundError を送出するヘルパー。"""
        status = self._orders.get(order_id)
        if status is None:
            raise OrderNotFoundError(f"Order not found: {order_id!r}")
        return status

📋 コピー# modules/gke-workload-identity/main.tf

# ── Google Service Account（GSA）──────────────────────────────────────────
resource "google_service_account" "order_api" {
  account_id   = "order-api-sa"
  display_name = "Order API — GKE Workload Identity SA"
  description  = "ECサイト注文APIがSecretManagerにアクセスするためのGSA"
  project      = var.project_id
}

# ── Secret Manager へのアクセス権限（最小権限）──────────────────────────────
# ノード SA ではなく、この GSA にのみ secretAccessor を付与
resource "google_project_iam_member" "order_api_secret_accessor" {
  project = var.project_id
  role    = "roles/secretmanager.secretAccessor"
  member  = "serviceAccount:${google_service_account.order_api.email}"
}

# ── Workload Identity バインディング（KSA → GSA）────────────────────────────
# K8s Pod（KSA）が GSA の権限を借用できるようにする
# principal 形式: serviceAccount:.svc.id.goog[/]
resource "google_service_account_iam_member" "workload_identity_binding" {
  service_account_id = google_service_account.order_api.name
  role               = "roles/iam.workloadIdentityUser"
  member             = "serviceAccount:${var.project_id}.svc.id.goog[${var.k8s_namespace}/${var.k8s_service_account_name}]"
}

# ── Secret Manager シークレット定義 ────────────────────────────────────────
resource "google_secret_manager_secret" "cloud_sql_dsn" {
  secret_id = "order-api-cloud-sql-dsn"
  project   = var.project_id

  replication {
    auto {}  # 自動レプリケーション（単一リージョン指定不要）
  }

  labels = {
    app         = "order-api"
    environment = var.environment
  }
}

resource "google_secret_manager_secret_version" "cloud_sql_dsn" {
  secret      = google_secret_manager_secret.cloud_sql_dsn.id
  secret_data = var.cloud_sql_dsn

  lifecycle {
    ignore_changes = [secret_data]  # CI/CD パイプライン側で更新させる
  }
}

📋 コピー# modules/gke-workload-identity/variables.tf

variable "project_id" {
  type        = string
  description = "GCP プロジェクト ID"
}

variable "environment" {
  type        = string
  description = "環境識別子（staging / production）"
  validation {
    condition     = contains(["staging", "production"], var.environment)
    error_message = "environment は staging または production のみ許可"
  }
}

variable "k8s_namespace" {
  type        = string
  description = "K8s Namespace（例: production）"
}

variable "k8s_service_account_name" {
  type        = string
  description = "K8s ServiceAccount 名（例: order-api-ksa）"
}

variable "cloud_sql_dsn" {
  type        = string
  sensitive   = true
  description = "Cloud SQL DSN（Secret Manager に格納する値）"
}