概要
StrEnum でオペレータを型安全に管理(Ch2)
FilterOperator(StrEnum) で "eq"・"in" などのマジックストリングを型に昇格。StrEnum は str と互換のため既存の文字列比較を壊さず、IDE の補完・参照追跡が効く。新オペレータ追加時の抜け漏れも静的に検知できる。
Annotated 型エイリアスで入力を宣言的に検証(Ch2)
FieldName = Annotated[str, Field(pattern=r"^[a-zA-Z_][a-zA-Z0-9_]*$")] で列名バリデーションを型の宣言に埋め込む。SQL インジェクション防止の正規表現が型定義に集約され、モデル外に散在するバリデーションコードが消える。
model_validator(mode='after') でフィールド間依存を検証(Ch3)
operator=in のとき values が必要というロジックを if/elif で各所に散在させず、Pydantic の @model_validator(mode='after') でモデル層に 1 箇所集約。mode='after' は全フィールド変換後に実行され、StrEnum 変換済みの self.operator を安全に参照できる。
Protocol + match 文で実装を差し替え可能に(Ch6/Ch8)
FilterRenderer(Protocol) で SQL レンダリングのインターフェースを定義。CampaignFilterBuilder は BigQueryFilterRenderer を直接 import しない。match 文でオペレータ別の SQL 生成を網羅的に記述し、case _: で未処理ケースを早期検知する。
問題
ECサイト MOps チームでは、キャンペーン配信の対象者フィルタリング条件を API 経由で受け取り、BigQuery クエリに変換するモジュールがある。以下の「悪いコード」は if/elif の連鎖・マジックストリング・未検証の入力・型安全性のない条件構築で実装されている。
問題点を全て洗い出し、Pydantic v2 の model_validator・@field_validator・Annotated 型制約・StrEnum・Protocol・match 文を使って Bad→Good にリファクタリングしてください。
制約・前提条件
- Python 3.12+、
pydantic>=2.7 StrEnumでフィルタ演算子(eq,neq,gt,gte,lt,lte,in,not_in)を定義Annotated[str, Field(min_length=1, max_length=64, pattern=r'^[a-zA-Z_][a-zA-Z0-9_]*$')]でフィールド名バリデーション@model_validator(mode='after')でin/not_in演算子使用時のvalues必須チェックmatch文でオペレータ別の SQL フラグメント生成(if/elif連鎖を排除)ProtocolでFilterRendererインターフェースを定義し、BigQuery 実装を差し替え可能にするfrozen=TrueでFilterConditionを値オブジェクト化- Google スタイル docstring・インラインコメント・名前付き定数を含めること
悪いコード (Before)
"""キャンペーン配信対象フィルタ → BigQuery WHERE 句変換(悪い実装)"""
# 問題1: operator を文字列で管理 — マジックストリング散在
VALID_OPERATORS = ["eq", "neq", "gt", "gte", "lt", "lte", "in", "not_in"]
def build_filter_clause(conditions: list[dict]) -> str:
"""条件リストから SQL WHERE フラグメントを生成する。"""
clauses = []
for cond in conditions:
field = cond.get("field")
operator = cond.get("operator")
value = cond.get("value")
values = cond.get("values")
# 問題2: 入力バリデーションなし — field に任意の文字列が入れられる(SQLインジェクション)
# 問題3: operator の網羅チェックなし — VALID_OPERATORS にない値がスルーされる
if operator not in VALID_OPERATORS:
raise ValueError(f"Invalid operator: {operator}")
# 問題4: if/elif の連鎖 — 新 operator 追加時に抜け漏れが起きやすい
if operator == "eq":
clause = f"{field} = '{value}'"
elif operator == "neq":
clause = f"{field} != '{value}'"
elif operator == "gt":
clause = f"{field} > {value}"
elif operator == "gte":
clause = f"{field} >= {value}"
elif operator == "lt":
clause = f"{field} < {value}"
elif operator == "lte":
clause = f"{field} <= {value}"
elif operator == "in":
# 問題5: in/not_in で values が None のとき TypeError がランタイムまで気づけない
vals = ", ".join(f"'{v}'" for v in values)
clause = f"{field} IN ({vals})"
elif operator == "not_in":
vals = ", ".join(f"'{v}'" for v in values)
clause = f"{field} NOT IN ({vals})"
else:
clause = ""
if clause:
clauses.append(clause)
return " AND ".join(clauses) if clauses else "TRUE"
class CampaignFilter:
# 問題6: BigQuery の SQL 生成ロジックが直接埋め込まれている(密結合 Ch8)
# 問題7: テスト時に SQL 実装をモックに差し替えられない(Ch6 依存性の注入なし)
def get_target_users_query(self, campaign_id: str, conditions: list[dict]) -> str:
where = build_filter_clause(conditions)
return (
f"SELECT user_id FROM `mops.members` "
f"WHERE {where} "
f"AND campaign_id = '{campaign_id}'"
)
# 使用例(壊れやすい)
filter_instance = CampaignFilter()
query = filter_instance.get_target_users_query(
campaign_id="CP-2026-SUMMER",
conditions=[
{"field": "age", "operator": "gte", "value": 20},
{"field": "rank", "operator": "in", "values": ["gold", "platinum"]},
{"field": "unsubscribed", "operator": "eq", "value": "false"},
]
)
print(query)
"eq", "in" 等の文字列が散在。class FilterOperator(StrEnum) で型安全に管理field に任意文字列が入り SQLインジェクション可能。Annotated[str, Field(pattern=r"^[a-zA-Z_][a-zA-Z0-9_]*$")] で厳格にバリデーションvalues=None のまま IN 句を生成すると TypeError がランタイムに発生。@model_validator(mode='after') でモデル生成時に即検知clause = "" で黙殺される。match 文 + case _: raise AssertionError で網羅性を保証FilterCondition(BaseModel, frozen=True) で型安全な値オブジェクトにCampaignFilter 全体を書き直す必要がある。BigQueryFilterRenderer として分離Protocol FilterRenderer を定義し、CampaignFilterBuilder.__init__(renderer) で注入可能にヒント(段階的開示)
ヒント1 — 方向性
if/elif でオペレータ文字列を比較しているコードを見たら、まず「網羅性が保証されているか」を疑う。match 文は case が全オペレータを網羅しないと case _: で黙殺される。StrEnum と組み合わせると IDE が未処理ケースを警告でき、新オペレータ追加時の抜け漏れを防げる。Pydantic v2 の @field_validator はフィールド単体の値検証、@model_validator(mode='after') はフィールド間の相互依存検証(例: operator=in のとき values が必要)に使い分ける。
ヒント2 — アプローチ
class FilterOperator(StrEnum): EQ = "eq"; NEQ = "neq"; GT = "gt"; GTE = "gte"; LT = "lt"; LTE = "lte"; IN = "in"; NOT_IN = "not_in"FieldName = Annotated[str, Field(min_length=1, max_length=64, pattern=r'^[a-zA-Z_][a-zA-Z0-9_]*$')]class FilterCondition(BaseModel): model_config = {"frozen": True}; field: FieldName; operator: FilterOperator; value: str | int | float | None = None; values: tuple[str | int | float, ...] | None = None@model_validator(mode="after")でoperator in {FilterOperator.IN, FilterOperator.NOT_IN}のときvaluesがNoneまたは空ならValueErrorclass FilterRenderer(Protocol): def render(self, condition: FilterCondition) -> str: ...class BigQueryFilterRenderer: def render(self, condition): match condition.operator: case FilterOperator.EQ: ...class CampaignFilterBuilder: def __init__(self, renderer: FilterRenderer): self._renderer = renderer
ヒント3 — コードの骨格
from enum import StrEnum
from typing import Annotated, Protocol
from pydantic import BaseModel, Field, model_validator
class FilterOperator(StrEnum):
EQ = "eq"
NEQ = "neq"
GT = "gt"
GTE = "gte"
LT = "lt"
LTE = "lte"
IN = "in"
NOT_IN = "not_in"
FieldName = Annotated[str, Field(min_length=1, max_length=64, pattern=r"^[a-zA-Z_][a-zA-Z0-9_]*$")]
class FilterCondition(BaseModel):
model_config = {"frozen": True}
field: FieldName
operator: FilterOperator
value: str | int | float | None = None
values: tuple[str | int | float, ...] | None = None
@model_validator(mode="after")
def _check_in_requires_values(self) -> "FilterCondition":
if self.operator in {FilterOperator.IN, FilterOperator.NOT_IN}:
if not self.values:
raise ValueError(f"operator='{self.operator}' requires non-empty 'values'")
return self
class FilterRenderer(Protocol):
def render(self, condition: FilterCondition) -> str: ...
class BigQueryFilterRenderer:
def render(self, condition: FilterCondition) -> str:
col = f"`{condition.field}`"
match condition.operator:
case FilterOperator.EQ: return f"{col} = {_quote(condition.value)}"
case FilterOperator.NEQ: return f"{col} != {_quote(condition.value)}"
case FilterOperator.GT: return f"{col} > {condition.value}"
case FilterOperator.GTE: return f"{col} >= {condition.value}"
case FilterOperator.LT: return f"{col} < {condition.value}"
case FilterOperator.LTE: return f"{col} <= {condition.value}"
case FilterOperator.IN:
vals = ", ".join(_quote(v) for v in (condition.values or ()))
return f"{col} IN ({vals})"
case FilterOperator.NOT_IN:
vals = ", ".join(_quote(v) for v in (condition.values or ()))
return f"{col} NOT IN ({vals})"
case _:
raise AssertionError(f"Unhandled: {condition.operator!r}")
class CampaignFilterBuilder:
def __init__(self, renderer: FilterRenderer) -> None:
self._renderer = renderer
def build(self, conditions: list[FilterCondition]) -> str:
if not conditions:
return "TRUE"
return " AND ".join(self._renderer.render(c) for c in conditions)
問題点分析(7点)
| # | 問題点 | 分類 | 改善方法 |
|---|---|---|---|
| 1 | オペレータがマジックストリング | 型安全 Ch2 | class FilterOperator(StrEnum) |
| 2 | フィールド名に入力バリデーションなし | セキュリティ Ch2 | Annotated[str, Field(pattern=r"^[a-zA-Z_][a-zA-Z0-9_]*$")] |
| 3 | in/not_in の values 必須チェックなし | 検証 Ch3 | @model_validator(mode='after') |
| 4 | if/elif の連鎖で網羅性未保証 | 条件分岐 Ch3 | match 文 + case _: raise AssertionError |
| 5 | 条件が dict — 型安全性ゼロ | 不変性 Ch2/Ch4 | FilterCondition(BaseModel, frozen=True) |
| 6 | BigQuery SQL 生成が密結合 | 密結合 Ch8 | BigQueryFilterRenderer として分離 |
| 7 | 依存性の注入なしでテスト不可 | 設計 Ch6 | Protocol FilterRenderer + コンストラクタ注入 |
設計図(SVG)
模範解答
# 問題1: operator がマジックストリング
VALID_OPERATORS = ["eq", "neq", "gt", "gte", "lt", "lte", "in", "not_in"]
def build_filter_clause(conditions: list[dict]) -> str:
clauses = []
for cond in conditions:
field = cond.get("field") # 問題2: 未検証(SQLインジェクション可能)
operator = cond.get("operator")
value = cond.get("value")
values = cond.get("values") # 問題3: None でも IN 句に流れ込む
if operator not in VALID_OPERATORS:
raise ValueError(f"Invalid operator: {operator}")
# 問題4: if/elif 連鎖 — 新 operator 追加時に抜け漏れ
if operator == "eq":
clause = f"{field} = '{value}'"
elif operator == "neq":
clause = f"{field} != '{value}'"
elif operator == "gt":
clause = f"{field} > {value}"
elif operator == "gte":
clause = f"{field} >= {value}"
elif operator == "lt":
clause = f"{field} < {value}"
elif operator == "lte":
clause = f"{field} <= {value}"
elif operator == "in":
vals = ", ".join(f"'{v}'" for v in values) # values=None → TypeError
clause = f"{field} IN ({vals})"
elif operator == "not_in":
vals = ", ".join(f"'{v}'" for v in values)
clause = f"{field} NOT IN ({vals})"
else:
clause = "" # 黙殺
if clause:
clauses.append(clause)
return " AND ".join(clauses) if clauses else "TRUE"
class CampaignFilter:
# 問題6: BigQuery 直接依存(密結合)
# 問題7: 依存性の注入なし(テスト不可)
def get_target_users_query(self, campaign_id, conditions):
where = build_filter_clause(conditions)
return (
f"SELECT user_id FROM `mops.members` "
f"WHERE {where} AND campaign_id = '{campaign_id}'"
)
"""campaign_filter.py — キャンペーン配信対象フィルタ条件 → SQL フラグメント変換
良いコード・悪いコードで学ぶ設計入門(改訂新版)
Ch2: StrEnum × Annotated 型エイリアス × 名前付き定数
Ch3: match 文 × model_validator でフィールド間依存を宣言的に検証
Ch6: Protocol + 依存性の注入
Ch8: BigQueryFilterRenderer として密結合を解消
"""
from __future__ import annotations
from enum import StrEnum
from typing import Annotated, Final, Protocol
from pydantic import BaseModel, Field, model_validator
# ── 定数 ──────────────────────────────────────────────────────────────
MAX_FIELD_NAME_LEN: Final[int] = 64
FIELD_NAME_PATTERN: Final[str] = r"^[a-zA-Z_][a-zA-Z0-9_]*$"
# ── StrEnum — オペレータ種別(Ch2)─────────────────────────────────────
class FilterOperator(StrEnum):
"""フィルタ演算子。StrEnum なので str と互換。"""
EQ = "eq" # 等値
NEQ = "neq" # 非等値
GT = "gt" # 超過
GTE = "gte" # 以上
LT = "lt" # 未満
LTE = "lte" # 以下
IN = "in" # リスト内包
NOT_IN = "not_in" # リスト非包含
# IN / NOT_IN 演算子セットを 1 箇所で管理
_LIST_OPS: Final[frozenset[FilterOperator]] = frozenset(
{FilterOperator.IN, FilterOperator.NOT_IN}
)
# ── Annotated 型エイリアス(Ch2)─────────────────────────────────────
# フィールド名: 英数字・アンダースコアのみ(SQL インジェクション防止)
FieldName = Annotated[
str,
Field(
min_length=1,
max_length=MAX_FIELD_NAME_LEN,
pattern=FIELD_NAME_PATTERN,
),
]
ScalarValue = str | int | float
# ── Pydantic v2 モデル(Ch2/Ch3/Ch4)────────────────────────────────
class FilterCondition(BaseModel):
"""単一フィルタ条件(値オブジェクト)。"""
model_config = {"frozen": True} # hashable + immutable(Ch4)
field: FieldName
operator: FilterOperator
value: ScalarValue | None = None
values: tuple[ScalarValue, ...] | None = None # listより hashable な tuple
@model_validator(mode="after")
def _validate_list_operator_has_values(self) -> "FilterCondition":
"""in / not_in 演算子使用時は values の非空を保証する(Ch3)。"""
if self.operator in _LIST_OPS:
if not self.values: # None または空タプル
raise ValueError(
f"operator='{self.operator}' requires non-empty 'values', "
f"but got: {self.values!r}"
)
return self
# ── Protocol — SQL レンダラーのインターフェース(Ch6/Ch8)────────────
class FilterRenderer(Protocol):
"""FilterCondition → SQL フラグメントへの変換インターフェース。"""
def render(self, condition: FilterCondition) -> str: ...
# ── BigQuery 実装──────────────────────────────────────────────────────
def _quote(value: ScalarValue) -> str:
"""値を BigQuery SQL リテラルに変換する(検証済み値のみ適用)。"""
if isinstance(value, str):
return f"'{value.replace(chr(39), chr(92) + chr(39))}'"
return str(value)
class BigQueryFilterRenderer:
"""BigQuery 標準 SQL 用 FilterRenderer 実装。"""
def render(self, condition: FilterCondition) -> str:
"""match 文でオペレータ別の SQL フラグメントを生成する(Ch3)。"""
col = f"`{condition.field}`" # バッククォートで列名を囲む
match condition.operator:
case FilterOperator.EQ:
return f"{col} = {_quote(condition.value)}" # type: ignore[arg-type]
case FilterOperator.NEQ:
return f"{col} != {_quote(condition.value)}" # type: ignore[arg-type]
case FilterOperator.GT:
return f"{col} > {condition.value}"
case FilterOperator.GTE:
return f"{col} >= {condition.value}"
case FilterOperator.LT:
return f"{col} < {condition.value}"
case FilterOperator.LTE:
return f"{col} <= {condition.value}"
case FilterOperator.IN:
vals = ", ".join(_quote(v) for v in condition.values) # type: ignore[union-attr]
return f"{col} IN ({vals})"
case FilterOperator.NOT_IN:
vals = ", ".join(_quote(v) for v in condition.values) # type: ignore[union-attr]
return f"{col} NOT IN ({vals})"
case _:
# 新 operator 追加時に render() 更新漏れを即検知
raise AssertionError(f"Unhandled operator: {condition.operator!r}")
# ── ビルダー(依存注入)(Ch6)────────────────────────────────────────
class CampaignFilterBuilder:
"""複数の FilterCondition を AND 結合した WHERE 句フラグメントを生成する。
Args:
renderer: FilterRenderer を満たす任意の実装(テスト時はモックに差し替え可)
"""
def __init__(self, renderer: FilterRenderer) -> None:
self._renderer = renderer
def build(self, conditions: list[FilterCondition]) -> str:
"""条件リストを AND 結合した SQL フラグメントに変換する。"""
if not conditions:
return "TRUE" # 条件なし → 全件対象
return " AND ".join(self._renderer.render(c) for c in conditions)
from pydantic import ValidationError
renderer = BigQueryFilterRenderer()
builder = CampaignFilterBuilder(renderer)
# 通常ケース
conditions = [
FilterCondition(field="age", operator=FilterOperator.GTE, value=20),
FilterCondition(field="member_rank", operator=FilterOperator.IN, values=("gold", "platinum")),
FilterCondition(field="last_purchase_days", operator=FilterOperator.LTE, value=90),
FilterCondition(field="unsubscribed", operator=FilterOperator.EQ, value="false"),
]
result = builder.build(conditions)
print("WHERE", result)
# WHERE `age` >= 20 AND `member_rank` IN ('gold', 'platinum')
# AND `last_purchase_days` <= 90 AND `unsubscribed` = 'false'
# 空条件
print(builder.build([]))
# TRUE
# バリデーションエラー: in に values なし
try:
FilterCondition(field="status", operator=FilterOperator.IN)
except ValidationError as e:
print(f"\nValidationError (in without values): {e.error_count()} error(s)")
# ValidationError (in without values): 1 error(s)
# バリデーションエラー: フィールド名が不正
try:
FilterCondition(field="1invalid_field", operator=FilterOperator.EQ, value="x")
except ValidationError as e:
print(f"ValidationError (bad field name): {e.error_count()} error(s)")
# ValidationError (bad field name): 1 error(s)
# frozen → 変更不可
cond = FilterCondition(field="age", operator=FilterOperator.EQ, value=30)
try:
cond.value = 999 # type: ignore
except Exception as e:
print(f"{type(e).__name__}: frozen=True のため変更不可")
# ValidationError: frozen=True のため変更不可
# StrEnum は str と互換
print(FilterOperator.EQ == "eq") # True
print(f"operator={FilterOperator.IN}") # operator=in
| ポイント | 適用した設計原則/パターン | 書籍対応章 |
|---|---|---|
FilterOperator(StrEnum) でオペレータを型安全に管理 | マジックストリング排除・型安全 | Ch2 |
Annotated FieldName で列名バリデーションを型に集約 | 型エイリアス・SQL インジェクション防止 | Ch2 |
FilterCondition(frozen=True) で値オブジェクト化 | 不変性・hashable(dict キー化可) | Ch4 |
@model_validator(mode='after') でフィールド間依存検証 | 条件分岐の複雑さを下げる・Fail-Fast | Ch3 |
match 文 + case _: raise AssertionError | 網羅性の保証・条件分岐の明確化 | Ch3 |
Protocol FilterRenderer で SQL 方言を差し替え可能に | 依存性の注入・密結合解消 | Ch6 / Ch8 |
CampaignFilterBuilder.__init__(renderer) | コンストラクタ注入・テスタビリティ向上 | Ch6 |
values: tuple[..., ...] で hashable + immutable | 不変性・型安全 | Ch4 |
# tests/test_campaign_filter.py
import pytest
from pydantic import ValidationError
from campaign_filter import (
FilterCondition, FilterOperator, FilterRenderer,
BigQueryFilterRenderer, CampaignFilterBuilder,
)
# ── FilterCondition ────────────────────────────────────────────────────
class TestFilterCondition:
def test_valid_eq(self):
c = FilterCondition(field="age", operator=FilterOperator.EQ, value=20)
assert c.field == "age"
assert c.operator == FilterOperator.EQ
def test_frozen(self):
c = FilterCondition(field="age", operator=FilterOperator.EQ, value=20)
with pytest.raises(Exception):
c.value = 999 # type: ignore
def test_in_requires_values(self):
with pytest.raises(ValidationError, match="requires non-empty 'values'"):
FilterCondition(field="rank", operator=FilterOperator.IN)
def test_not_in_requires_values(self):
with pytest.raises(ValidationError):
FilterCondition(field="rank", operator=FilterOperator.NOT_IN, values=())
def test_in_with_values_valid(self):
c = FilterCondition(field="rank", operator=FilterOperator.IN, values=("gold",))
assert c.values == ("gold",)
def test_invalid_field_name(self):
with pytest.raises(ValidationError):
FilterCondition(field="1bad", operator=FilterOperator.EQ, value="x")
def test_field_name_with_space(self):
with pytest.raises(ValidationError):
FilterCondition(field="bad field", operator=FilterOperator.EQ, value="x")
# ── BigQueryFilterRenderer ─────────────────────────────────────────────
class TestBigQueryFilterRenderer:
def setup_method(self):
self.r = BigQueryFilterRenderer()
def render(self, **kw):
return self.r.render(FilterCondition(**kw))
def test_eq_string(self):
sql = self.render(field="status", operator=FilterOperator.EQ, value="active")
assert sql == "`status` = 'active'"
def test_gte_int(self):
sql = self.render(field="age", operator=FilterOperator.GTE, value=20)
assert sql == "`age` >= 20"
def test_in_strings(self):
sql = self.render(field="rank", operator=FilterOperator.IN, values=("gold", "platinum"))
assert sql == "`rank` IN ('gold', 'platinum')"
def test_not_in(self):
sql = self.render(field="rank", operator=FilterOperator.NOT_IN, values=("bronze",))
assert sql == "`rank` NOT IN ('bronze')"
def test_backtick_column(self):
sql = self.render(field="order_count", operator=FilterOperator.GT, value=0)
assert sql.startswith("`order_count`")
# ── CampaignFilterBuilder ──────────────────────────────────────────────
class MockRenderer:
"""テスト用モックレンダラー(Protocol 実装)。"""
def render(self, condition: FilterCondition) -> str:
return f"[{condition.field}]"
class TestCampaignFilterBuilder:
def setup_method(self):
self.builder = CampaignFilterBuilder(MockRenderer())
def test_empty_conditions(self):
assert self.builder.build([]) == "TRUE"
def test_single_condition(self):
c = FilterCondition(field="age", operator=FilterOperator.EQ, value=20)
assert self.builder.build([c]) == "[age]"
def test_multiple_and_join(self):
conds = [
FilterCondition(field="age", operator=FilterOperator.EQ, value=20),
FilterCondition(field="rank", operator=FilterOperator.EQ, value="gold"),
]
assert self.builder.build(conds) == "[age] AND [rank]"
def test_accepts_protocol_impl(self):
# Protocol 実装なら何でも注入できることを確認
real_builder = CampaignFilterBuilder(BigQueryFilterRenderer())
c = FilterCondition(field="age", operator=FilterOperator.LTE, value=65)
result = real_builder.build([c])
assert "`age` <= 65" == result
ポイント解説
"eq" などのマジックストリングを FilterOperator.EQ に昇格。StrEnum は str と互換のため既存の文字列比較を壊さず、IDE の補完・参照追跡が効く。FieldName = Annotated[str, Field(pattern=...)] はバリデーションロジックを型の宣言に埋め込み、SQL インジェクション防止の正規表現が型定義に集約される。operator=in のとき values が必要というロジックを if/elif で各所に散在させず、モデル層で 1 箇所に集約。mode='after' は全フィールドの変換後に実行されるため、StrEnum 変換済みの self.operator を安全に参照できる。if/elif の連鎖は新 case 追加時の抜け漏れに気づけない。match + case _: raise AssertionError で開発時に即検知できる。frozen=True で hashable・immutable にし、キャッシュキーとして安全に使える。CampaignFilterBuilder は BigQueryFilterRenderer を直接 import しない。テスト時に MockRenderer を注入でき、将来の Redshift / DuckDB 対応も差し替えのみで実現できる。実務への応用
ECサイト MOps のキャンペーン配信では、Pub/Sub → Cloud Run v2 で受け取った JSON ペイロードをそのまま BigQuery の WHERE 句に変換するケースが多い。入力が外部(フロントエンド・マーケター操作 UI)から来る場合、FieldName の正規表現バリデーションが SQL インジェクションの第一防御線になる(内部コードは信頼するが、外部入力は必ずバリデーション)。
Protocol FilterRenderer を使うと pytest で MockRenderer を注入してレンダリング結果をアサートするユニットテストが書きやすくなる。BigQueryFilterRenderer の実際の SQL が BigQuery で正しく解釈されるかは統合テスト層で検証する。
Argo Workflows の CronWorkflow でバッチ配信フィルタを再利用するときも、FilterCondition が frozen=True で hashable なため、処理済みキャッシュの key として使える。
FilterCondition を FastAPI のリクエストボディとして直接使うことができる(Pydantic v2 は FastAPI v0.104+ とネイティブ統合)。model_config = {"frozen": True} でも FastAPI のシリアライズは動作する。
今日のまとめ
次のステップ
- 発展問題:
FilterConditionのOR結合・ネスト(FilterGroup)を実装し、match文で再帰的に処理する問題。Pydantic のmodel_rebuild()と循環参照解決も絡める。 - 参考: Pydantic v2 ドキュメント
model_validator・Annotated/ Python 3.10match文 PEP 634 / 「良いコード・悪いコードで学ぶ設計入門(改訂新版)」Ch3・Ch6・Ch8