A コーディング — Pydantic v2 model_validator(mode='after') × Annotated FieldName × StrEnum FilterOperator × match 文 SQL フラグメント生成 × Protocol FilterRenderer 依存注入(キャンペーン対象フィルタ Bad→Good Ch2/Ch3/Ch6/Ch8)

2026-07-06 (Day 95) 月曜 コーディング ★★★★☆ Python 3.12 / Pydantic v2 / StrEnum / match 文 / Protocol 良いコード設計入門 Ch2 / Ch3 / Ch6 / Ch8

概要

🏷️

StrEnum でオペレータを型安全に管理(Ch2)

FilterOperator(StrEnum)"eq""in" などのマジックストリングを型に昇格。StrEnumstr と互換のため既存の文字列比較を壊さず、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 レンダリングのインターフェースを定義。CampaignFilterBuilderBigQueryFilterRenderer を直接 import しない。match 文でオペレータ別の SQL 生成を網羅的に記述し、case _: で未処理ケースを早期検知する。

問題

ECサイト MOps チームでは、キャンペーン配信の対象者フィルタリング条件を API 経由で受け取り、BigQuery クエリに変換するモジュールがある。以下の「悪いコード」は if/elif の連鎖・マジックストリング・未検証の入力・型安全性のない条件構築で実装されている。

問題点を全て洗い出し、Pydantic v2 の model_validator@field_validatorAnnotated 型制約・StrEnumProtocolmatch 文を使って 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 連鎖を排除)
  • ProtocolFilterRenderer インターフェースを定義し、BigQuery 実装を差し替え可能にする
  • frozen=TrueFilterCondition を値オブジェクト化
  • Google スタイル docstring・インラインコメント・名前付き定数を含めること
期待する回答形式: 問題点の列挙(番号付き)+ 改善後コード(Google スタイル docstring・インラインコメント・名前付き定数含む)+ 実行例(input→output)+ 適用した設計パターン名と書籍対応章

悪いコード (Before)

このコードには 7つの設計上の問題 が隠れています。見つけてみてください。
bad_campaign_filter.py — if/elif + マジックストリング + 未検証 + 密結合
"""キャンペーン配信対象フィルタ → 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)
問題点サマリー(7点)
1オペレータがマジックストリング(Ch2)"eq", "in" 等の文字列が散在。class FilterOperator(StrEnum) で型安全に管理
2フィールド名に入力バリデーションなし(Ch2/セキュリティ)field に任意文字列が入り SQLインジェクション可能。Annotated[str, Field(pattern=r"^[a-zA-Z_][a-zA-Z0-9_]*$")] で厳格にバリデーション
3in/not_in の values 必須チェックなし(Ch3)values=None のまま IN 句を生成すると TypeError がランタイムに発生。@model_validator(mode='after') でモデル生成時に即検知
4if/elif の連鎖(Ch3) — 新 operator 追加時に elif 分岐を追加し忘れると clause = "" で黙殺される。match 文 + case _: raise AssertionError で網羅性を保証
5条件が dict — 型安全性ゼロ(Ch2/Ch4) — キーのタイポを型チェッカーが検出できない。FilterCondition(BaseModel, frozen=True) で型安全な値オブジェクトに
6BigQuery SQL 生成が直接埋め込まれている(Ch8: 密結合) — SQL 方言が変わると CampaignFilter 全体を書き直す必要がある。BigQueryFilterRenderer として分離
7テスト時に実装を差し替えられない(Ch6: 依存性の注入なし)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} のとき valuesNone または空なら ValueError
  • class 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オペレータがマジックストリング型安全 Ch2class FilterOperator(StrEnum)
2フィールド名に入力バリデーションなしセキュリティ Ch2Annotated[str, Field(pattern=r"^[a-zA-Z_][a-zA-Z0-9_]*$")]
3in/not_in の values 必須チェックなし検証 Ch3@model_validator(mode='after')
4if/elif の連鎖で網羅性未保証条件分岐 Ch3match 文 + case _: raise AssertionError
5条件が dict — 型安全性ゼロ不変性 Ch2/Ch4FilterCondition(BaseModel, frozen=True)
6BigQuery SQL 生成が密結合密結合 Ch8BigQueryFilterRenderer として分離
7依存性の注入なしでテスト不可設計 Ch6Protocol FilterRenderer + コンストラクタ注入

設計図(SVG)

External API JSON ペイロード parse FilterCondition BaseModel, frozen=True field: FieldName (Annotated) operator: FilterOperator value: str | int | float | None values: tuple[...] | None @model_validator(mode='after') FilterOperator (StrEnum) EQ / NEQ / GT / GTE / LT / LTE / IN / NOT_IN str 互換・IDE 補完・網羅チェック可 FilterRenderer Protocol render(condition) → str BigQueryFilterRenderer implements FilterRenderer match condition.operator: case EQ / NEQ / ... / IN / _ implements CampaignFilterBuilder __init__(renderer: FilterRenderer) build(conditions) → str uses BigQuery SQL `age` >= 20 AND ... 凡例 Protocol(インターフェース) Pydantic BaseModel(frozen=True) StrEnum(型安全な定数)

模範解答

Before — if/elif + dict + マジックストリング + 密結合
# 問題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}'"
        )
After — StrEnum × Annotated × model_validator × match × Protocol
"""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-FastCh3
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

ポイント解説

1StrEnum でオペレータを型安全に管理(Ch2: 名前付き定数・型エイリアス)"eq" などのマジックストリングを FilterOperator.EQ に昇格。StrEnumstr と互換のため既存の文字列比較を壊さず、IDE の補完・参照追跡が効く。
2Annotated 型エイリアスで入力バリデーションを型定義に集約(Ch2: 型エイリアス)FieldName = Annotated[str, Field(pattern=...)] はバリデーションロジックを型の宣言に埋め込み、SQL インジェクション防止の正規表現が型定義に集約される。
3model_validator(mode='after') でフィールド間依存を宣言的に検証(Ch3: 条件分岐の複雑さ)operator=in のとき values が必要というロジックを if/elif で各所に散在させず、モデル層で 1 箇所に集約。mode='after' は全フィールドの変換後に実行されるため、StrEnum 変換済みの self.operator を安全に参照できる。
4match 文でオペレータ別の SQL 生成を網羅的に記述(Ch3: 条件分岐)if/elif の連鎖は新 case 追加時の抜け漏れに気づけない。match + case _: raise AssertionError で開発時に即検知できる。
5frozen=True で FilterCondition を値オブジェクト化(Ch4: 不変性) — 条件オブジェクトが生成後に変更されると SQL フラグメントとの整合性が崩れる。frozen=True で hashable・immutable にし、キャッシュキーとして安全に使える。
6Protocol で FilterRenderer インターフェースを定義(Ch6/Ch8: 依存性の注入・密結合解消)CampaignFilterBuilderBigQueryFilterRenderer を直接 import しない。テスト時に MockRenderer を注入でき、将来の Redshift / DuckDB 対応も差し替えのみで実現できる。

実務への応用

ECサイト MOps のキャンペーン配信では、Pub/Sub → Cloud Run v2 で受け取った JSON ペイロードをそのまま BigQuery の WHERE 句に変換するケースが多い。入力が外部(フロントエンド・マーケター操作 UI)から来る場合、FieldName の正規表現バリデーションが SQL インジェクションの第一防御線になる(内部コードは信頼するが、外部入力は必ずバリデーション)。

Protocol FilterRenderer を使うと pytestMockRenderer を注入してレンダリング結果をアサートするユニットテストが書きやすくなる。BigQueryFilterRenderer の実際の SQL が BigQuery で正しく解釈されるかは統合テスト層で検証する。

Argo Workflows の CronWorkflow でバッチ配信フィルタを再利用するときも、FilterConditionfrozen=True で hashable なため、処理済みキャッシュの key として使える。

Cloud Run v2 × Pydantic v2 構成: Cloud Run v2 に FastAPI をデプロイする場合、FilterCondition を FastAPI のリクエストボディとして直接使うことができる(Pydantic v2 は FastAPI v0.104+ とネイティブ統合)。model_config = {"frozen": True} でも FastAPI のシリアライズは動作する。

今日のまとめ

StrEnum + match 文でオペレータ分岐を網羅的に表現し、Pydantic の @model_validator でフィールド間依存を型安全に検証することで、if/elif・マジックストリング・手動バリデーション散在という3つの問題を一気に解消できる。Protocol で実装を差し替え可能にしておくことで、SQL 方言変更・テストが低コストになる。

次のステップ

  • 発展問題: FilterConditionOR 結合・ネスト(FilterGroup)を実装し、match 文で再帰的に処理する問題。Pydantic の model_rebuild() と循環参照解決も絡める。
  • 参考: Pydantic v2 ドキュメント model_validatorAnnotated / Python 3.10 match 文 PEP 634 / 「良いコード・悪いコードで学ぶ設計入門(改訂新版)」Ch3・Ch6・Ch8

自己評価(あとで記入)