A+B 複合 — Generic[T] × Ok[T]/Err × ParamSpec × TypeIs(型安全 Result 型 Bad→Good Ch2/Ch4/Ch10)× GKE Gateway API × HTTPRoute × ReferenceGrant × GCPManagedCertificate × SSL Policy MODERN(Terraform)

2026-07-04 (Day 93) 土曜複合問題 ★★★★☆ Python 3.12 / Generic[T] / ParamSpec / TypeIs / Ok[T] / Err / frozen dataclass GKE Gateway API / HTTPRoute / ReferenceGrant / GCPManagedCertificate / SSL Policy

概要

🔷

Generic[T] + Ok[T]/Err で型安全 Result 型

Generic[T] を継承することで __class_getitem__ が自動提供され、Ok[int] のような型パラメータ記法が使える。Ok[T](成功)と Err(失敗)の 2 サブクラス分離により、result.value の型が T として静的に推論され、mypy/pyright が型エラーを検出できる。

📝

ParamSpec で高階関数の引数型を完全保持

P = ParamSpec("P")T = TypeVar("T")Callable[P, T] を受け取ると、*args: P.args, **kwargs: P.kwargs が元の関数の引数型をそのまま伝播する。run_job(int, "42")int("42") と同じ型シグネチャで推論され、引数の型ミスを静的に検出できる。

🔍

TypeIs[Ok[T]] — 双方向ナローイング(PEP 742)

TypeIs[T]TypeGuard[T] と異なり else ブランチでも型が絞り込まれる。if is_ok(result):else ブランチで resultErr として推論されるため、result.message に安全にアクセスできる(双方向ナローイング)。

🌐

GKE Gateway API — Ingress からの移行と役割分離

Gateway API(gateway.networking.k8s.io/v1)は Ingress の後継。Gateway(インフラチーム管理・LB 設定)と HTTPRoute(アプリチーム管理・ルーティング)を分離できる。ReferenceGrant でクロス namespace 参照を宣言的に許可し、GCPManagedCertificate で TLS 証明書を自動管理する。

問題 A: コーディング — Generic[T] × ParamSpec × TypeIs(型安全 Result 型 Bad→Good)

以下の「悪いコード」は、ECサイト MOps チームのバッチジョブ結果を扱う汎用コンテナです。問題点を全て洗い出し、Generic[T]・Ok[T]/Err サブクラス分離・ParamSpec・TypeIs(PEP 742)・frozen dataclass + slots=True を使って Bad→Good にリファクタリングしてください。

制約・前提条件

  • Python 3.12+(TypeIstyping.TypeIsParamSpectyping.ParamSpec
  • Result[T]Generic[T] で実装し、__class_getitem__Generic 継承で自動提供されることを理解すること
  • Ok[T] / Err の 2 サブクラスで成功/失敗を型レベルで分離すること(@dataclass(frozen=True, slots=True)
  • run_jobParamSpec + TypeVar で引数・戻り値の型を完全に保持すること
  • TypeIs[Ok[T]] 型ガードを実装し、if is_ok(result):result.valueT として推論されること
  • Google スタイル docstring・インラインコメント・名前付き定数を含めること
期待する回答形式: 問題点の列挙(番号付き)+ 改善後コード + 実行例(input→output)+ 適用した設計パターン名と書籍対応章

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

このコードには 7つの設計上の問題 が隠れています。見つけてみてください。
bad_result.py — 型なし汎用コンテナ・握りつぶし・型ガードなし
class Result:
    def __init__(self, ok, value=None, error=None):
        self.ok = ok
        self.value = value  # 問題1: Any型(型推論不可)
        self.error = error  # 問題2: 同一クラスで成功/失敗を区別

def run_job(fn, *args):          # 問題3: fn と args の型なし
    try:
        return Result(ok=True, value=fn(*args))
    except Exception as e:
        return Result(ok=False, error=str(e))  # 問題4: exc を str に変換して握りつぶし

# 問題5: 戻り値型が Result(型パラメータなし)
res = run_job(lambda x: x * 2, 10)

# 問題6: value の型が分からない(Any → mypy エラー検出不可)
print(res.value + 1)   # mypy はエラーを検出できない

# 問題7: if res.ok: は型ナローイングにならない
if res.ok:
    # res.value が何型か mypy には不明(Any のまま)
    print(res.value)
else:
    # res.error が str なのか Exception なのか不明
    print(res.error)
問題点サマリー(7点)
1Result クラスに型パラメータなし(Ch2)self.valueAny のため mypy が res.value + 1 のエラーを検出できない。Generic[T] で型パラメータ化する
2成功/失敗が同一クラスで区別不可(Ch2/Ch4)ok: bool フラグで区別するのは型不安全。Ok[T] / Err の 2 サブクラスに分離し Union 型でパターンマッチする
3run_job の引数型がなし(Ch2)fn*argsAnyParamSpec で関数シグネチャを完全に型保持する
4エラーを str(e) で握りつぶし(Ch10) — 元の例外が失われてスタックトレースを再取得できない。Errexc: Exception | None を持たせる
5戻り値が Result(型パラメータなし)(Ch2) — 呼び出し側で res.value の型が不明。Ok[T] | Err を返す型アノテーションで型推論を有効化
6value と error が同時に None 可能(Ch10) — バリデーションなし。Err は必ず message: str を持ち Ok は必ず value: T を持つ構造に変更
7型ガードなし(Ch2)if res.ok: は型ナローイングにならない。TypeIs[Ok[T]] 型ガードで if is_ok(res): ブランチで res.valueT として推論されるようにする

ヒント A(段階的開示)

ヒント1 — 方向性
Generic[T] を継承することで __class_getitem__ が自動的に利用可能になる(Python 3.9+)。Ok[T] は成功値を T として保持し、Err はエラーメッセージと例外オブジェクトを保持する。run_jobP = ParamSpec("P")T = TypeVar("T")Callable[P, T] を受け取り、戻り値を Ok[T] | Err として返す。TypeIs[Ok[T]] 型ガードにより if is_ok(result): ブランチで result.valueT として推論される。else ブランチでは resultErr として推論される(双方向ナローイング)。
ヒント2 — アプローチ
  • T = TypeVar("T")P = ParamSpec("P") を定義
  • @dataclass(frozen=True, slots=True) class Ok(Generic[T]):value: T を持たせる
  • @dataclass(frozen=True, slots=True) class Err:message: strexc: Exception | None = None を持たせる
  • type Result[T] = Ok[T] | Err(PEP 695 型エイリアス)を定義
  • def run_job(fn: Callable[P, T], *args: P.args, **kwargs: P.kwargs) -> Ok[T] | Err: で型安全な実行ラッパー
  • def is_ok(result: Ok[T] | Err) -> TypeIs[Ok[T]]: で双方向型ガード
  • def collect_results(results: list[Ok[T] | Err]) -> tuple[list[T], list[Err]]: で成功/失敗を分離
ヒント3 — コードの骨格
from __future__ import annotations
from dataclasses import dataclass
from typing import Callable, Generic, ParamSpec, TypeIs, TypeVar

T = TypeVar("T")
P = ParamSpec("P")

@dataclass(frozen=True, slots=True)
class Ok(Generic[T]):
    value: T  # 型 T として保持(Generic[T] により型パラメータ化)

@dataclass(frozen=True, slots=True)
class Err:
    message: str
    exc: Exception | None = None  # スタックトレース保持(握りつぶし禁止)

# PEP 695 型エイリアス(Python 3.12+)
type Result[T] = Ok[T] | Err

def is_ok(result: Ok[T] | Err) -> TypeIs[Ok[T]]:
    """双方向ナローイング: else ブランチで Err として推論される。"""
    return isinstance(result, Ok)

def run_job(fn: Callable[P, T], *args: P.args, **kwargs: P.kwargs) -> Ok[T] | Err:
    try:
        return Ok(value=fn(*args, **kwargs))
    except Exception as exc:
        return Err(message=f"{fn.__name__} 失敗: {exc}", exc=exc)

問題点分析 — カテゴリ A

#問題点分類改善方法
1Result クラスに型パラメータなし型の活用 Ch2Generic[T] で型パラメータ化
2成功/失敗が同一クラスで区別不可型の活用 Ch2/Ch4Ok[T] / Err の 2 サブクラス分離
3run_job の引数型がなし型の活用 Ch2ParamSpec で関数シグネチャを完全保持
4エラーを str(e) で握りつぶしエラー処理 Ch10exc: Exception | None を保持
5戻り値が Result(型パラメータなし)型の活用 Ch2Ok[T] | Err を返す型アノテーション
6value と error が同時に None 可能エラー処理 Ch102 クラス分離で構造的に排除
7型ガードなし(ナローイング不可)型の活用 Ch2TypeIs[Ok[T]] 双方向型ガード(PEP 742)

模範解答 A

Before — 型なし・Ok/Err 混在・握りつぶし・型ガードなし
class Result:
    def __init__(self, ok, value=None, error=None):
        self.ok = ok
        self.value = value  # Any
        self.error = error  # str or None(型不明)

def run_job(fn, *args):      # 引数型なし
    try:
        return Result(ok=True, value=fn(*args))
    except Exception as e:
        return Result(ok=False, error=str(e))  # 握りつぶし

res = run_job(lambda x: x * 2, 10)
print(res.value + 1)   # mypy: エラー検出不可(Any)

if res.ok:              # 型ナローイングにならない
    print(res.value)   # mypy: Any のまま
After — Generic[T] × Ok[T]/Err × ParamSpec × TypeIs
"""result.py — Generic[T] × ParamSpec × TypeIs(Ch2/Ch4/Ch10)"""
from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Callable, Final, Generic, ParamSpec, TypeIs, TypeVar

logger = logging.getLogger(__name__)

T = TypeVar("T")
P = ParamSpec("P")

MAX_ERROR_MESSAGE_LEN: Final[int] = 1024


@dataclass(frozen=True, slots=True)
class Ok(Generic[T]):
    """成功を表す値オブジェクト。value は型 T として保持される。

    Args:
        value: 成功時の返却値。型 T として静的型チェッカーに公開される。
    """
    value: T  # 型 T として保持(Generic[T] により型パラメータ化)


@dataclass(frozen=True, slots=True)
class Err:
    """失敗を表す値オブジェクト。スタックトレースを exc に保持する。

    Args:
        message: 人間可読なエラーメッセージ。
        exc: 元の例外オブジェクト(ログ/再 raise 用)。
    """
    message: str
    exc: Exception | None = None  # スタックトレース保持(握りつぶし禁止)

    def __post_init__(self) -> None:
        if len(self.message) > MAX_ERROR_MESSAGE_LEN:
            object.__setattr__(
                self, "message",
                self.message[:MAX_ERROR_MESSAGE_LEN] + "..."
            )


# PEP 695 型エイリアス(Python 3.12+)
type Result[T] = Ok[T] | Err


def is_ok(result: Ok[T] | Err) -> TypeIs[Ok[T]]:
    """成功型ガード(双方向ナローイング・PEP 742)。

    if ブランチ: result.value が T として推論される。
    else ブランチ: result が Err として推論される(双方向性)。
    """
    return isinstance(result, Ok)


def run_job(
    fn: Callable[P, T],
    *args: P.args,
    **kwargs: P.kwargs,
) -> Ok[T] | Err:
    """型安全なジョブ実行ラッパー。

    fn の引数シグネチャ(P)と戻り値型(T)を完全に保持する。
    成功時は Ok[T]、例外発生時は Err を返す(例外を握りつぶさない)。
    """
    try:
        result_value = fn(*args, **kwargs)  # P.args/P.kwargs で型安全
        logger.debug("run_job 成功: fn=%s", fn.__name__)
        return Ok(value=result_value)       # Ok[T] を返す
    except Exception as exc:               # 例外を Err に変換(握りつぶし禁止)
        message = f"{fn.__name__} 失敗: {exc}"
        logger.warning(
            "run_job 失敗: fn=%s error=%s", fn.__name__, exc, exc_info=True
        )
        return Err(message=message, exc=exc)  # exc を保持


def collect_results(results: list[Ok[T] | Err]) -> tuple[list[T], list[Err]]:
    """Ok/Err を分離して返す(SRP による責務分離)。"""
    oks: list[T] = []
    errs: list[Err] = []
    for r in results:
        if is_ok(r):
            oks.append(r.value)  # TypeIs により T として推論
        else:
            errs.append(r)       # TypeIs 双方向: Err として推論
    return oks, errs
# 正常系
res1: Ok[int] | Err = run_job(int, "42")
if is_ok(res1):
    val: int = res1.value   # mypy: int として正確に推論(Any ではない)
    print(val + 1)          # → 43

# 異常系(例外を Err に変換)
res2: Ok[int] | Err = run_job(int, "not_a_number")
if is_ok(res2):
    print(res2.value)
else:
    # TypeIs の双方向ナローイング: ここでは res2 が Err として推論される
    print(res2.message)     # → "int 失敗: invalid literal for int()..."
    print(type(res2.exc))   # → <class 'ValueError'>(握りつぶし禁止)

# 複数ジョブの集計
jobs = [run_job(int, s) for s in ["1", "2", "bad", "4"]]
oks, errs = collect_results(jobs)
print(oks)       # → [1, 2, 4]
print(len(errs)) # → 1

# Ok は frozen=True で変更不可(値オブジェクトの不変性)
res3 = run_job(str.upper, "hello")
if is_ok(res3):
    try:
        res3.value = "world"  # type: ignore[misc]
    except Exception:
        print("FrozenInstanceError: slots=True で変更不可")

# PEP 695 型エイリアス
r: Result[str] = Ok(value="hello")
print(isinstance(r, Ok))  # → True

# ParamSpec 型安全確認
def fetch(url: str, timeout: int) -> bytes:
    return b"response"

res4: Ok[bytes] | Err = run_job(fetch, "https://example.com", 30)
# run_job(fetch, "https://example.com", "not_int")  # mypy: 型エラー(P 保持)
ポイント適用した設計原則/パターン書籍対応章
Ok[T] / Err 2 サブクラス分離型の活用・値オブジェクトCh2/Ch4
Generic[T] + frozen dataclass不変性・型パラメータ化Ch2/Ch4
ParamSpec で引数型を完全保持型の活用・高階関数Ch2
TypeIs[Ok[T]] 双方向ナローイング型ガード・PEP 742Ch2
Errexc 保持(握りつぶし禁止)エラー処理・Fail-LoudCh10
__post_init__ バリデーションエラー処理Ch10
collect_results で責務分離SRP・単一責任Ch6
# tests/test_result.py
import pytest
from result import Ok, Err, run_job, collect_results, is_ok, MAX_ERROR_MESSAGE_LEN


class TestOk:
    def test_value_stored(self):
        r = Ok(value=42)
        assert r.value == 42

    def test_frozen_immutability(self):
        r = Ok(value="hello")
        with pytest.raises(Exception):  # FrozenInstanceError
            r.value = "world"  # type: ignore[misc]

    def test_generic_int(self):
        r: Ok[int] = Ok(value=100)
        assert isinstance(r.value, int)


class TestErr:
    def test_message_stored(self):
        e = Err(message="失敗", exc=None)
        assert e.message == "失敗"

    def test_exc_stored(self):
        exc = ValueError("test error")
        e = Err(message="エラー", exc=exc)
        assert e.exc is exc  # 元の例外オブジェクトが保持されること

    def test_message_truncated(self):
        long_msg = "x" * (MAX_ERROR_MESSAGE_LEN + 100)
        e = Err(message=long_msg)
        assert len(e.message) == MAX_ERROR_MESSAGE_LEN + 3  # "..." が追加される

    def test_frozen_immutability(self):
        e = Err(message="エラー")
        with pytest.raises(Exception):
            e.message = "変更"  # type: ignore[misc]


class TestIsOk:
    def test_ok_returns_true(self):
        r: Ok[int] | Err = Ok(value=1)
        assert is_ok(r) is True

    def test_err_returns_false(self):
        r: Ok[int] | Err = Err(message="失敗")
        assert is_ok(r) is False

    def test_bidirectional_narrowing(self):
        """TypeIs の双方向ナローイングのテスト。"""
        r: Ok[str] | Err = Ok(value="hello")
        if is_ok(r):
            # if ブランチ: Ok[str] として推論される
            assert r.value == "hello"
        else:
            # else ブランチ: Err として推論される(このテストでは通らない)
            assert isinstance(r, Err)


class TestRunJob:
    def test_success_returns_ok(self):
        res = run_job(int, "42")
        assert is_ok(res)
        assert isinstance(res, Ok)
        if is_ok(res):
            assert res.value == 42

    def test_failure_returns_err(self):
        res = run_job(int, "not_a_number")
        assert not is_ok(res)
        assert isinstance(res, Err)

    def test_err_preserves_exc(self):
        res = run_job(int, "bad")
        if not is_ok(res):
            assert res.exc is not None
            assert isinstance(res.exc, ValueError)

    def test_paramspec_preserves_types(self):
        def add(x: int, y: int) -> int:
            return x + y

        res = run_job(add, 3, 4)
        if is_ok(res):
            assert res.value == 7


class TestCollectResults:
    def test_separates_ok_and_err(self):
        results: list[Ok[int] | Err] = [
            run_job(int, "1"),
            run_job(int, "bad"),
            run_job(int, "3"),
        ]
        oks, errs = collect_results(results)
        assert oks == [1, 3]
        assert len(errs) == 1

    def test_all_ok(self):
        results: list[Ok[int] | Err] = [Ok(1), Ok(2), Ok(3)]
        oks, errs = collect_results(results)
        assert oks == [1, 2, 3]
        assert errs == []

    def test_all_err(self):
        results: list[Ok[int] | Err] = [
            Err(message="err1"), Err(message="err2")
        ]
        oks, errs = collect_results(results)
        assert oks == []
        assert len(errs) == 2

問題 B: インフラ — GKE Gateway API × HTTPRoute × ReferenceGrant × GCPManagedCertificate(Terraform)

ECサイト MOps チームの API 基盤が Ingress(NGINX コントローラー)で構築されており、以下の課題を抱えています。

現状の課題:
  • networking.k8s.io/v1 Ingress + NGINX コントローラーを使用 → Gateway API(gateway.networking.k8s.io/v1)への移行が必要
  • HTTPS 証明書が手動更新(Let's Encrypt certbot cron)→ GCPManagedCertificate で自動化が必要
  • MOps API と決済 API が同一 namespace で動作し、HTTPRoute のクロス namespace 参照に ReferenceGrant が必要
  • TLS ポリシー(最小 TLS バージョン・暗号スイート)が未設定(TLS 1.0/1.1 使用可能)

要件

#要件
1gateway.networking.k8s.io/v1 Gateway(GatewayClass: gke-l7-global-external-managed)を Terraform で作成すること
2HTTPRoute でパスベースルーティング(/api/mops/* → MOps API、/api/payment/* → 決済 API)を設定すること
3クロス namespace 参照(Gateway は gateway-ns、MOps API は mops namespace)に ReferenceGrant を設定すること
4GCPManagedCertificate で TLS 証明書を自動管理し、SSL Policy MODERN(TLS 1.2+ のみ)を設定すること

ヒント B(段階的開示)

ヒント1 — 方向性
GKE Gateway API の GatewayClass は GKE がマネージドで提供(gke-l7-global-external-managed)するため自前で作成不要。Gateway リソースは networking.gke.io/managed-certificates アノテーションで GCPManagedCertificatenetworking.gke.io/v1 ManagedCertificate)を参照する。クロス namespace 参照は ReferenceGrant で「どの namespace の誰が参照できるか」を明示的に許可する。重要: ReferenceGrant は参照「される」側の namespace に配置する(mops namespace に配置して gateway-ns からの参照を許可する)。
ヒント2 — Terraform リソース構成
  • kubernetes_manifest(Terraform Kubernetes provider)で Gateway / HTTPRoute / ReferenceGrant / ManagedCertificate を管理
  • google_compute_ssl_policy(profile = "MODERN", min_tls_version = "TLS_1_2")で TLS ポリシー
  • networking.gke.io/v1beta1 FrontendConfig で HTTP → HTTPS リダイレクトと SSL Policy 参照
  • HTTPRoutefiltersResponseHeaderModifier(HSTS ヘッダー追加)
  • ReferenceGrantfromGatewayHTTPRoute の両方を指定(セキュリティ)
ヒント3 — Gateway + ReferenceGrant の骨格(YAML)
Gateway リソース
apiVersion: gateway.networking.k8s.io/v1
kind: Gateway
metadata:
  name: mops-gateway
  namespace: gateway-ns
  annotations:
    networking.gke.io/managed-certificates: "mops-cert"
    networking.gke.io/v1beta1.FrontendConfig: "mops-frontend-config"
spec:
  gatewayClassName: gke-l7-global-external-managed
  listeners:
  - name: https
    port: 443
    protocol: HTTPS
    tls:
      mode: Terminate
      certificateRefs:
      - group: networking.gke.io
        kind: ManagedCertificate
        name: mops-cert
    allowedRoutes:
      namespaces:
        from: All
ReferenceGrant(参照される側の namespace に配置)
# mops namespace に配置して
# gateway-ns からの参照を許可する
apiVersion: gateway.networking.k8s.io/v1beta1
kind: ReferenceGrant
metadata:
  name: allow-gateway-ns
  namespace: mops  # ← 参照「される」側
spec:
  from:
  - group: gateway.networking.k8s.io
    kind: Gateway
    namespace: gateway-ns  # ← 参照「する」側
  to:
  - group: ""
    kind: Service

アーキテクチャ図 — GKE Gateway API × HTTPRoute × ReferenceGrant × GCPManagedCertificate

クライアント HTTPS :443 HTTP :80 → Redirect Google Cloud / GKE Autopilot GCPManagedCertificate + SSL Policy ManagedCertificate networking.gke.io/v1 自動更新・Google 管理 SSL Policy MODERN min_tls_version: TLS_1_2 TLS 1.0/1.1 を拒否 FrontendConfig networking.gke.io/v1beta1 HTTP → HTTPS リダイレクト 301 Moved Permanently sslPolicy: mops-ssl-policy Namespace: gateway-ns(インフラチーム管理) Gateway — mops-gateway(GatewayClass: gke-l7-global-external-managed) listener: http :80 → FrontendConfig で Redirect allowedRoutes: namespaces: All listener: https :443 tls.mode: Terminate certificateRefs: mops-cert annotations managed-certificates: mops-cert FrontendConfig: mops-frontend-config Namespace: mops(MOps チーム管理) ReferenceGrant allow-gateway-ns from: gateway-ns/Gateway to: mops/Service ← 参照される側に配置(重要) HTTPRoute mops-api-route /api/mops/* → mops-api-svc filter: HSTS + X-Content-Type parentRefs: gateway-ns/mops-gateway mops-api-svc :8080 → mops-api Pod(GKE Autopilot) Deployment: mops-api / replicas: auto(HPA) Namespace: payment(決済チーム管理) ReferenceGrant allow-gateway-ns from: gateway-ns/Gateway to: payment/Service ← 参照される側に配置(重要) HTTPRoute payment-api-route /api/payment/* → payment-svc hostnames: api.payment.example.com parentRefs: gateway-ns/mops-gateway payment-api-svc :8080 → payment-api Pod(GKE Autopilot) PCI DSS スコープ / 独立 namespace でアクセス制御 Terraform 管理領域(terraform/modules/mops-gateway/) kubernetes_manifest (Gateway) kubernetes_manifest (HTTPRoute x2) kubernetes_manifest (ReferenceGrant x2) kubernetes_manifest (ManagedCertificate) google_compute_ssl_policy(MODERN / TLS_1_2)+ kubernetes_namespace(gateway-ns / mops / payment) HTTPS :443 TLS 証明書 HTTP→HTTPS ReferenceGrant 許可後ルーティング ReferenceGrant 許可後ルーティング

模範解答 B

# terraform/modules/mops-gateway/main.tf
# GKE Gateway API × HTTPRoute × ReferenceGrant × GCPManagedCertificate

locals {
  project_id     = var.project_id
  region         = var.region
  domain_mops    = "api.mops.example.com"
  domain_payment = "api.payment.example.com"
}

# ── 1. GCPManagedCertificate — TLS 証明書の自動管理 ─────────────────────────────
# Before: certbot cron で手動更新(更新失敗リスク・監視必要)
# After: Google が自動プロビジョニング・更新(zero-ops)
resource "kubernetes_manifest" "mops_cert" {
  manifest = {
    apiVersion = "networking.gke.io/v1"
    kind       = "ManagedCertificate"
    metadata = {
      name      = "mops-cert"
      namespace = "gateway-ns"
    }
    spec = {
      # 両ドメインを 1 つの ManagedCertificate で管理(SAN 証明書)
      domains = [local.domain_mops, local.domain_payment]
    }
  }
}

# ── 2. SSL Policy — TLS 1.2+ / 推奨暗号スイートのみ ────────────────────────────
# Before: SSL ポリシー未設定(TLS 1.0/1.1・弱い暗号が使用可能)
# After: MODERN プロファイルで TLS 1.2+ のみ許可(PCI DSS 準拠)
resource "google_compute_ssl_policy" "mops_ssl_policy" {
  name            = "mops-ssl-policy"
  project         = local.project_id
  profile         = "MODERN"     # ECDHE-ECDSA-AES128-GCM-SHA256 等の推奨スイートのみ
  min_tls_version = "TLS_1_2"   # TLS 1.0/1.1 を拒否
}

# ── 3. FrontendConfig — HTTP → HTTPS リダイレクト ────────────────────────────────
resource "kubernetes_manifest" "mops_frontend_config" {
  manifest = {
    apiVersion = "networking.gke.io/v1beta1"
    kind       = "FrontendConfig"
    metadata = {
      name      = "mops-frontend-config"
      namespace = "gateway-ns"
    }
    spec = {
      redirectToHttps = {
        enabled          = true
        responseCodeName = "MOVED_PERMANENTLY_DEFAULT"  # 301 リダイレクト
      }
      sslPolicy = google_compute_ssl_policy.mops_ssl_policy.name
    }
  }
  depends_on = [google_compute_ssl_policy.mops_ssl_policy]
}

# ── 4. Gateway — GKE L7 External Load Balancer ──────────────────────────────────
# Before: NGINX Ingress(コントローラー自前運用・アップグレード負荷)
# After: GKE マネージド GatewayClass(インフラを Google が管理)
resource "kubernetes_manifest" "mops_gateway" {
  manifest = {
    apiVersion = "gateway.networking.k8s.io/v1"
    kind       = "Gateway"
    metadata = {
      name      = "mops-gateway"
      namespace = "gateway-ns"
      annotations = {
        # GCPManagedCertificate を参照
        "networking.gke.io/managed-certificates"   = "mops-cert"
        # FrontendConfig で HTTP→HTTPS リダイレクト + SSL Policy を適用
        "networking.gke.io/v1beta1.FrontendConfig" = "mops-frontend-config"
      }
    }
    spec = {
      # GKE が提供する GatewayClass(自前で GatewayClass を作成不要)
      gatewayClassName = "gke-l7-global-external-managed"
      listeners = [
        # HTTP :80 → FrontendConfig で 301 リダイレクト
        {
          name     = "http"
          port     = 80
          protocol = "HTTP"
          allowedRoutes = { namespaces = { from = "All" } }
        },
        # HTTPS :443 — TLS を Gateway で終端(バックエンドは HTTP)
        {
          name     = "https"
          port     = 443
          protocol = "HTTPS"
          tls = {
            mode = "Terminate"  # TLS を Gateway で終端
            certificateRefs = [
              {
                group = "networking.gke.io"
                kind  = "ManagedCertificate"
                name  = "mops-cert"  # GCPManagedCertificate を参照
              }
            ]
          }
          # 全 namespace からの HTTPRoute アタッチを許可(ReferenceGrant で更に絞る)
          allowedRoutes = { namespaces = { from = "All" } }
        }
      ]
    }
  }
  depends_on = [
    kubernetes_manifest.mops_cert,
    kubernetes_manifest.mops_frontend_config,
  ]
}

# ── 5. ReferenceGrant — クロス namespace 参照の明示的許可 ────────────────────────
# Before: Ingress は namespace スコープ → クロス ns 参照不可(同一 ns に同居)
# After: ReferenceGrant で「参照される側」が許可を宣言(セキュリティ向上)
#
# 重要: ReferenceGrant は参照「される」側の namespace に配置する

resource "kubernetes_manifest" "mops_reference_grant" {
  manifest = {
    apiVersion = "gateway.networking.k8s.io/v1beta1"
    kind       = "ReferenceGrant"
    metadata = {
      name      = "allow-gateway-ns"
      namespace = "mops"  # ← 参照「される」側の namespace に配置
    }
    spec = {
      from = [
        {
          group     = "gateway.networking.k8s.io"
          kind      = "Gateway"
          namespace = "gateway-ns"  # ← 参照「する」側
        }
      ]
      to = [
        {
          group = ""      # core API group(Service)
          kind  = "Service"  # mops namespace の Service への参照を許可
        }
      ]
    }
  }
}

resource "kubernetes_manifest" "payment_reference_grant" {
  manifest = {
    apiVersion = "gateway.networking.k8s.io/v1beta1"
    kind       = "ReferenceGrant"
    metadata = {
      name      = "allow-gateway-ns"
      namespace = "payment"  # ← 決済チームが自分の namespace で許可を管理
    }
    spec = {
      from = [
        {
          group     = "gateway.networking.k8s.io"
          kind      = "Gateway"
          namespace = "gateway-ns"
        }
      ]
      to = [{ group = ""; kind = "Service" }]
    }
  }
}

# ── 6. HTTPRoute — MOps API パスルーティング(mops namespace)────────────────────
resource "kubernetes_manifest" "mops_api_route" {
  manifest = {
    apiVersion = "gateway.networking.k8s.io/v1"
    kind       = "HTTPRoute"
    metadata = {
      name      = "mops-api-route"
      namespace = "mops"  # MOps チームが自分の namespace で管理
    }
    spec = {
      parentRefs = [
        {
          name        = "mops-gateway"
          namespace   = "gateway-ns"  # クロス ns 参照(ReferenceGrant が許可)
          sectionName = "https"       # HTTPS listener のみに限定
        }
      ]
      hostnames = [local.domain_mops]
      rules = [
        {
          matches = [{ path = { type = "PathPrefix"; value = "/api/mops" } }]
          backendRefs = [
            {
              name      = "mops-api-svc"
              port      = 8080
              namespace = "mops"
              weight    = 100
            }
          ]
          # セキュリティヘッダーをレスポンスに追加
          filters = [
            {
              type = "ResponseHeaderModifier"
              responseHeaderModifier = {
                add = [
                  {
                    name  = "Strict-Transport-Security"
                    value = "max-age=31536000; includeSubDomains; preload"  # HSTS
                  },
                  {
                    name  = "X-Content-Type-Options"
                    value = "nosniff"
                  },
                  {
                    name  = "X-Frame-Options"
                    value = "DENY"
                  }
                ]
              }
            }
          ]
        }
      ]
    }
  }
  depends_on = [
    kubernetes_manifest.mops_gateway,
    kubernetes_manifest.mops_reference_grant,
  ]
}

# ── 7. HTTPRoute — 決済 API(payment namespace)────────────────────────────────
resource "kubernetes_manifest" "payment_api_route" {
  manifest = {
    apiVersion = "gateway.networking.k8s.io/v1"
    kind       = "HTTPRoute"
    metadata = {
      name      = "payment-api-route"
      namespace = "payment"  # 決済チームが自分の namespace で管理(PCI DSS スコープ分離)
    }
    spec = {
      parentRefs = [
        {
          name        = "mops-gateway"
          namespace   = "gateway-ns"
          sectionName = "https"
        }
      ]
      hostnames = [local.domain_payment]
      rules = [
        {
          matches = [{ path = { type = "PathPrefix"; value = "/api/payment" } }]
          backendRefs = [
            {
              name      = "payment-api-svc"
              port      = 8080
              namespace = "payment"
              weight    = 100
            }
          ]
        }
      ]
    }
  }
  depends_on = [
    kubernetes_manifest.mops_gateway,
    kubernetes_manifest.payment_reference_grant,
  ]
}

Bad vs Good 設計比較

観点Bad(Ingress NGINX)Good(Gateway API)
TLS 証明書管理 certbot cron 手動更新(更新失敗リスク・監視必要) GCPManagedCertificate で自動更新(zero-ops)
TLS ポリシー 未設定(TLS 1.0/1.1・弱い暗号スイートが使用可能) SSL Policy MODERN(TLS 1.2+ のみ・ECDHE のみ)
Namespace 分離 MOps / 決済 API が同一 namespace に同居(RBAC 粒度が粗い) Gateway: gateway-ns / MOps: mops / 決済: payment に分離
クロス namespace 参照 Ingress は namespace スコープのため不可 ReferenceGrant で参照される側が明示的に許可を宣言
役割分離 インフラチームが全ルーティングを管理(ボトルネック) Gateway(インフラ管理)/ HTTPRoute(各チーム管理)で分離
セキュリティヘッダー Ingress では HSTS 等のヘッダー追加が annotation 依存 HTTPRoute filter で ResponseHeaderModifier(標準 API)
HTTP→HTTPS リダイレクト NGINX annotation で設定(GKE 版 NGINX 依存) FrontendConfig で標準化(GKE ネイティブ)
コントローラー運用 NGINX コントローラーを自前でアップグレード・監視 GKE マネージド GatewayClass(Google が管理)

確認コマンド(kubectl / gcloud)

# 1. Gateway の状態確認
kubectl get gateway mops-gateway -n gateway-ns -o jsonpath='{.status.conditions}' | jq .
# Expected: type="Programmed" status="True" が含まれれば L7 LB プロビジョニング済み
# NG: type="Accepted" が False の場合は GatewayClass 未検出(GKE バージョン確認)

# 2. ManagedCertificate の証明書状態確認
kubectl get managedcertificate mops-cert -n gateway-ns \
  -o jsonpath='{.status.certificateStatus}'
# Expected: "Active"(プロビジョニングには 15〜60 分かかる場合がある)
# Provisioning: DNS 伝播待ち / FailedNotVisible: DNS 設定を確認

# 3. HTTPRoute のルーティング確認
kubectl get httproute -A
# mops/mops-api-route と payment/payment-api-route が表示されること
# STATUS: Accepted であること

# 4. ReferenceGrant の確認
kubectl get referencegrant -n mops
kubectl get referencegrant -n payment
# allow-gateway-ns が各 namespace で表示されること

# 5. Gateway の IP アドレス取得
kubectl get gateway mops-gateway -n gateway-ns \
  -o jsonpath='{.status.addresses[0].value}'
# 払い出された External IP を DNS の A レコードに登録する

# 6. エンドポイント確認(証明書有効化後)
curl -I https://api.mops.example.com/api/mops/health
# Expected:
#   HTTP/2 200
#   strict-transport-security: max-age=31536000; includeSubDomains; preload
#   x-content-type-options: nosniff

# 7. HTTP → HTTPS リダイレクト確認
curl -I http://api.mops.example.com/api/mops/health
# Expected: HTTP/1.1 301 Moved Permanently
#           location: https://api.mops.example.com/api/mops/health

# 8. SSL ポリシー確認(TLS 1.1 が拒否されること)
openssl s_client -connect api.mops.example.com:443 -tls1_1 2>&1 | grep -i "alert\|error"
# Expected: "ssl alert" または "handshake failure"(TLS 1.1 拒否)

# 9. SSL ポリシーの Terraform 状態確認
gcloud compute ssl-policies describe mops-ssl-policy \
  --project=${PROJECT_ID} \
  --format="json" | jq '{profile: .profile, minTlsVersion: .minTlsVersion}'
# Expected: {"profile": "MODERN", "minTlsVersion": "TLS_1_2"}

# 10. ReferenceGrant なしでのアクセス拒否確認
# (テスト用に ReferenceGrant を削除して動作確認)
kubectl delete referencegrant allow-gateway-ns -n mops
kubectl get httproute mops-api-route -n mops -o jsonpath='{.status.parents}'
# Expected: type="ResolvedRefs" status="False" reason="RefNotPermitted"
# → ReferenceGrant がないと HTTPRoute がバックエンドを解決できないことを確認

ポイント解説

カテゴリ A

1 Generic[T] と __class_getitem__ の関係(Ch2)
Python 3.9 以降、Generic[T] を継承したクラスは __class_getitem__ が自動的に提供され、Ok[int] のような型パラメータ記法が使える。__class_getitem__ を手動実装する必要があるのは Generic を継承しない場合(例: list[int] を模倣したいが Generic を使わない場合)のみ。frozen=True, slots=True と組み合わせることで不変かつメモリ効率の高い値オブジェクトになる(slots=True__dict__ が省略されメモリ使用量が削減される)。
2 ParamSpec で関数シグネチャを完全保持(Ch2)
T = TypeVar("T") + P = ParamSpec("P")Callable[P, T] を受け取ると、*args: P.args, **kwargs: P.kwargs が元の関数の引数型をそのまま伝播する。run_job(int, "42")int("42") と同じ型シグネチャで推論され、引数の型ミスを静的に検出できる。デコレータパターン(@retry, @with_tracing)でも同様に使うと、デコレートされた関数の引数型が壊れない。
3 TypeIs[T] vs TypeGuard[T] の双方向ナローイング(Ch2)
TypeIs[T](PEP 742)は else ブランチでも型が絞り込まれる。if is_ok(result):else ブランチで resultErr として推論されるため、result.message に安全にアクセスできる。TypeGuard[T]if ブランチのみナローイング(else では元の型のまま)。実運用では typing_extensions.TypeIs で Python 3.12 にバックポートして使う。

カテゴリ B

4 Gateway API が Ingress より優れている3つの理由
(1) 役割分離 — インフラチームが Gateway(LB 設定)を管理し、アプリチームが HTTPRoute(ルーティング)を管理できる。Ingress では全てのルーティングが Ingress リソース 1 つに集中するため、インフラチームが全変更のボトルネックになる。(2) クロス namespace 参照ReferenceGrant で許可した組み合わせのみ許可され、セキュリティポリシーを宣言的に管理できる。(3) 機能の標準化 — TLS 設定・ヘッダー操作・リダイレクトが標準 API で記述でき、NGINX annotation の独自実装から脱却できる。
5 ReferenceGrant のセキュリティ設計意図(最小権限原則の namespace レベル実現)
ReferenceGrant は「参照される側の namespace」に配置する。これにより、外部 namespace から勝手に自分の Service を参照されることを防ぐ。gateway-ns の Gateway が mops の Service を参照するには、mops の管理者が ReferenceGrant を明示的に許可する必要がある。これは IAM の roles/iam.serviceAccountTokenCreator による impersonate 許可と同じ「許可は許可される側が与える」というセキュリティ原則を namespace レベルで実現したメカニズム。
6 GCPManagedCertificate + SSL Policy の組み合わせ(実務ポイント)
GCPManagedCertificate はドメイン所有確認→発行→自動更新を Google が管理するため、certbot の cron ジョブや更新失敗監視が不要になる。証明書の有効期限切れによる障害リスクを構造的に排除できる。SSL Policy MODERN は PCI DSS / OWASP TLS Cheat Sheet 準拠の暗号スイートのみを有効化し、TLS 1.0/1.1 と弱い暗号(RC4・DES・3DES)を無効化する。証明書プロビジョニングには 15〜60 分かかるため、初回デプロイ時は kubectl get managedcertificate -w でステータスを監視する。

実務への応用

  • Result[T] 型は MOps バッチパイプラインの標準エラー処理パターンになる: Argo Workflows の各ステップが Ok[int] | Err を返し、collect_results で成功/失敗を分離してから Slack 通知・リドライブ判定に使う設計にすることで、例外の握りつぶしとサイレント失敗を構造的に排除できる
  • ParamSpec はデコレータ設計で特に威力を発揮する: @retry@with_tracing のようなデコレータを ParamSpec で実装すると、デコレートされた関数の引数型が完全に保持され、呼び出し側の型チェックが壊れない。DataDog OTel span 追加デコレータを型安全に実装できる
  • Gateway API × ReferenceGrant は「チーム境界をコードで表現する」: ECサイトでは MOps チーム・決済チーム・在庫チームが同じ GKE クラスタを共有することが多い。Gateway を中央インフラチームが管理し、各チームが自分の namespace に HTTPRoute と ReferenceGrant を配置することで、Terraform の PR ベースレビューでルーティング変更を制御できる
  • GCPManagedCertificate の DNS 依存に注意: ドメインの DNS A レコードが Gateway の外部 IP を向いていないと ManagedCertificate のプロビジョニングが FailedNotVisible で止まる。Terraform で google_dns_record_setkubernetes_manifest(Gateway)` の depends_on 順序を正しく設定することが重要
  • 証券マン視点 — Gateway API 移行の ROI: Ingress NGINX の運用コスト(certbot 更新失敗対応: 月 2〜4h × ¥5,000/h = ¥10,000〜20,000 + NGINX コントローラーアップグレード: 四半期 4h = ¥20,000)と Gateway API への移行工数(2〜3 人日 = ¥150,000〜225,000)を比較すると、6〜12 ヶ月でペイバックできる投資案件として EM に提案できる

今日のまとめ

Generic[T] + Ok[T]/Err 分離 + ParamSpec + TypeIs[Ok[T]](PEP 742 双方向ナローイング)の組み合わせで「型安全 Result 型」を実装することで、Python バッチジョブのエラー処理が型レベルで保証される(Ch2/Ch4/Ch10)。if is_ok(result):result.valueT として、else ブランチで result.messagestr として mypy/pyright に推論され、例外の握りつぶしとサイレント失敗を構造的に排除できる。

インフラ側では GKE Gateway API(gateway.networking.k8s.io/v1 Gateway + HTTPRoute)への Ingress NGINX 移行により、インフラチーム(Gateway 管理)とアプリチーム(HTTPRoute 管理)の役割分離を実現し、ReferenceGrant でクロス namespace 参照を宣言的に制御する(「許可は許可される側が与える」最小権限原則)。GCPManagedCertificate で TLS 証明書を zero-ops 自動更新し、SSL Policy MODERN(TLS 1.2+)で PCI DSS 準拠のセキュア API ゲートウェイを Terraform で構築できる。

自己評価

自分の回答

気づき・メモ