概要
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 ブランチで result が Err として推論されるため、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+(
TypeIsはtyping.TypeIs、ParamSpecはtyping.ParamSpec) Result[T]をGeneric[T]で実装し、__class_getitem__はGeneric継承で自動提供されることを理解することOk[T]/Errの 2 サブクラスで成功/失敗を型レベルで分離すること(@dataclass(frozen=True, slots=True))run_jobはParamSpec + TypeVarで引数・戻り値の型を完全に保持することTypeIs[Ok[T]]型ガードを実装し、if is_ok(result):でresult.valueがTとして推論されること- Google スタイル docstring・インラインコメント・名前付き定数を含めること
悪いコード (Before) — カテゴリ A
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)
self.value が Any のため mypy が res.value + 1 のエラーを検出できない。Generic[T] で型パラメータ化するok: bool フラグで区別するのは型不安全。Ok[T] / Err の 2 サブクラスに分離し Union 型でパターンマッチするfn と *args が Any。ParamSpec で関数シグネチャを完全に型保持するErr に exc: Exception | None を持たせるres.value の型が不明。Ok[T] | Err を返す型アノテーションで型推論を有効化Err は必ず message: str を持ち Ok は必ず value: T を持つ構造に変更if res.ok: は型ナローイングにならない。TypeIs[Ok[T]] 型ガードで if is_ok(res): ブランチで res.value が T として推論されるようにするヒント A(段階的開示)
ヒント1 — 方向性
Generic[T] を継承することで __class_getitem__ が自動的に利用可能になる(Python 3.9+)。Ok[T] は成功値を T として保持し、Err はエラーメッセージと例外オブジェクトを保持する。run_job は P = ParamSpec("P") と T = TypeVar("T") で Callable[P, T] を受け取り、戻り値を Ok[T] | Err として返す。TypeIs[Ok[T]] 型ガードにより if is_ok(result): ブランチで result.value が T として推論される。else ブランチでは result が Err として推論される(双方向ナローイング)。
ヒント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: strとexc: 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
| # | 問題点 | 分類 | 改善方法 |
|---|---|---|---|
| 1 | Result クラスに型パラメータなし | 型の活用 Ch2 | Generic[T] で型パラメータ化 |
| 2 | 成功/失敗が同一クラスで区別不可 | 型の活用 Ch2/Ch4 | Ok[T] / Err の 2 サブクラス分離 |
| 3 | run_job の引数型がなし | 型の活用 Ch2 | ParamSpec で関数シグネチャを完全保持 |
| 4 | エラーを str(e) で握りつぶし | エラー処理 Ch10 | exc: Exception | None を保持 |
| 5 | 戻り値が Result(型パラメータなし) | 型の活用 Ch2 | Ok[T] | Err を返す型アノテーション |
| 6 | value と error が同時に None 可能 | エラー処理 Ch10 | 2 クラス分離で構造的に排除 |
| 7 | 型ガードなし(ナローイング不可) | 型の活用 Ch2 | TypeIs[Ok[T]] 双方向型ガード(PEP 742) |
模範解答 A
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 のまま
"""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 742 | Ch2 |
Err に exc 保持(握りつぶし禁止) | エラー処理・Fail-Loud | Ch10 |
__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 使用可能)
要件
| # | 要件 |
|---|---|
| 1 | gateway.networking.k8s.io/v1 Gateway(GatewayClass: gke-l7-global-external-managed)を Terraform で作成すること |
| 2 | HTTPRoute でパスベースルーティング(/api/mops/* → MOps API、/api/payment/* → 決済 API)を設定すること |
| 3 | クロス namespace 参照(Gateway は gateway-ns、MOps API は mops namespace)に ReferenceGrant を設定すること |
| 4 | GCPManagedCertificate で TLS 証明書を自動管理し、SSL Policy MODERN(TLS 1.2+ のみ)を設定すること |
ヒント B(段階的開示)
ヒント1 — 方向性
GatewayClass は GKE がマネージドで提供(gke-l7-global-external-managed)するため自前で作成不要。Gateway リソースは networking.gke.io/managed-certificates アノテーションで GCPManagedCertificate(networking.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 参照HTTPRouteのfiltersでResponseHeaderModifier(HSTS ヘッダー追加)ReferenceGrantのfromにGatewayとHTTPRouteの両方を指定(セキュリティ)
ヒント3 — Gateway + ReferenceGrant の骨格(YAML)
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
# 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
模範解答 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
Python 3.9 以降、
Generic[T] を継承したクラスは __class_getitem__ が自動的に提供され、Ok[int] のような型パラメータ記法が使える。__class_getitem__ を手動実装する必要があるのは Generic を継承しない場合(例: list[int] を模倣したいが Generic を使わない場合)のみ。frozen=True, slots=True と組み合わせることで不変かつメモリ効率の高い値オブジェクトになる(slots=True で __dict__ が省略されメモリ使用量が削減される)。
T = TypeVar("T") + P = ParamSpec("P") で Callable[P, T] を受け取ると、*args: P.args, **kwargs: P.kwargs が元の関数の引数型をそのまま伝播する。run_job(int, "42") は int("42") と同じ型シグネチャで推論され、引数の型ミスを静的に検出できる。デコレータパターン(@retry, @with_tracing)でも同様に使うと、デコレートされた関数の引数型が壊れない。
TypeIs[T](PEP 742)は else ブランチでも型が絞り込まれる。if is_ok(result): の else ブランチで result が Err として推論されるため、result.message に安全にアクセスできる。TypeGuard[T] は if ブランチのみナローイング(else では元の型のまま)。実運用では typing_extensions.TypeIs で Python 3.12 にバックポートして使う。
カテゴリ B
(1) 役割分離 — インフラチームが
Gateway(LB 設定)を管理し、アプリチームが HTTPRoute(ルーティング)を管理できる。Ingress では全てのルーティングが Ingress リソース 1 つに集中するため、インフラチームが全変更のボトルネックになる。(2) クロス namespace 参照 — ReferenceGrant で許可した組み合わせのみ許可され、セキュリティポリシーを宣言的に管理できる。(3) 機能の標準化 — TLS 設定・ヘッダー操作・リダイレクトが標準 API で記述でき、NGINX annotation の独自実装から脱却できる。
ReferenceGrant は「参照される側の namespace」に配置する。これにより、外部 namespace から勝手に自分の Service を参照されることを防ぐ。gateway-ns の Gateway が mops の Service を参照するには、mops の管理者が ReferenceGrant を明示的に許可する必要がある。これは IAM の roles/iam.serviceAccountTokenCreator による impersonate 許可と同じ「許可は許可される側が与える」というセキュリティ原則を namespace レベルで実現したメカニズム。
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_set→kubernetes_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.value が T として、else ブランチで result.message が str として 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 で構築できる。