A コーディング — コマンド・クエリ分離(CQS)× フラグ引数廃止 × null返却禁止 × 専用例外階層(available_qty 付き)× Protocol StockRepository 依存注入(フラッシュセール在庫予約 Bad→Good Ch4/Ch10/Ch13)

2026-07-13 (Day 99) 月曜 コーディング ★★★★☆ Python 3.12 / dataclass(frozen, slots) / Protocol / 例外階層 良いコード設計入門 Ch4 / Ch10 / Ch13

概要

🔀

コマンド・クエリ分離(CQS)の徹底(Ch13)

is_available()(クエリ・副作用なし)と reserve()(コマンド・副作用あり)を別クラスに分割。呼び出し側はメソッド名を見ただけで「確認だけ」か「在庫を実際に変更するか」を判断できる。

🚫

フラグ引数の廃止(Ch13)

dry_run / notify という2つのブールフラグを削除し、それぞれ別メソッド(is_available / reserve)・別クラス(ReservationNotifier)に分解。呼んでみるまで挙動が分からない曖昧さを排除する。

null返却禁止 → 専用例外で失敗を表現(Ch13/Ch10)

reserve() は成功時は必ず ReservationResult を返し、失敗時は必ず SkuNotFoundError / InsufficientStockError を送出する。InsufficientStockError.available_qty で失敗理由の文脈情報も保持する。

🔌

Protocol による依存性注入

StockRepository(Protocol) を定義し、本番は Cloud SQL/Spanner 実装、テストは InMemoryStockRepository に差し替え可能にする。値オブジェクトは frozen=True, slots=True(Ch4)で不変化。

問題

ECサイト MOps チームでは、フラッシュセール実施時に「在庫予約(在庫引当)」を行う InventoryService を Python で実装している。フロントエンドのカート画面から「在庫確認だけ(決済前のリアルタイム表示)」と「実際に予約を確定する(決済確定時)」の両方が同じメソッドを介して呼び出されており、以下の「悪いコード」になっている。

問題点を全て洗い出し、「良いコード・悪いコードで学ぶ設計入門」第13章(メソッド設計)の コマンド・クエリ分離(CQS)・フラグ引数の廃止・null返却禁止 を適用して Bad→Good にリファクタリングしてください。

制約・前提条件

  • Python 3.12+、dataclass(frozen=True, slots=True) を値オブジェクトに使用すること
  • 「在庫確認」(副作用なし・クエリ)と「在庫予約の確定」(副作用あり・コマンド)を別メソッド・別クラスに分離すること
  • 在庫不足・SKU未存在は None を返さず、専用の例外クラス階層(StockError を基底)で表現すること。InsufficientStockError には available_qty 属性を持たせること
  • notify(Slack通知)はコマンドメソッドから分離し、呼び出し側または別サービスの責務とすること
  • DBアクセスは ProtocolStockRepository インターフェースを定義し、依存性注入でテスト容易性を確保すること
  • Google スタイル docstring・インラインコメント・名前付き定数を含めること
期待する回答形式: 問題点の列挙(番号付き)+ 改善後コード(Google スタイル docstring・インラインコメント・名前付き定数含む)+ 実行例(input→output)+ 適用した設計パターン名と書籍対応章

悪いコード (Before)

このコードには 6つの設計上の問題 が隠れています。見つけてみてください。
bad_inventory.py — フラグ引数 × CQS違反 × null返却
class InventoryService:
    def __init__(self, db):
        self.db = db

    # 問題1・2: dry_run フラグで「確認」と「予約確定」を切り替え(CQS違反・フラグ引数)
    # 問題5: notify フラグも同居(通知という別の関心事が混入)
    def process_reservation(self, sku, qty, dry_run=False, notify=False):
        row = self.db.get_stock(sku)

        # 問題3・6: SKU未存在・在庫不足を区別できず None に潰す
        if row is None:
            return None
        current_qty = row["quantity"]
        if current_qty < qty:
            return None

        if dry_run:
            # 問題4: dry_run 時とそうでない時で戻り値のキーが違う(available vs reserved)
            return {"sku": sku, "available": True, "requested_qty": qty}

        row["quantity"] = current_qty - qty
        self.db.save_stock(sku, row)
        if notify:
            send_slack_notification(f"{sku} を {qty} 個予約しました")
        return {"sku": sku, "reserved": True, "remaining_qty": row["quantity"]}


def handle_flash_sale_order(service, sku, qty):
    result = service.process_reservation(sku, qty)
    # 呼び出し側で None チェックを忘れると…
    if result["reserved"]:  # ← result が None なら TypeError
        print(f"予約成功: {sku}")
問題点サマリー(6点)
1フラグ引数 dry_run(Ch13) — 「確認」と「予約確定」という別意図を1メソッドの引数値で切り替えている。呼び出し側は引数を見るまで副作用の有無が分からない
2CQS違反(Ch13) — クエリ(副作用なしの確認)とコマンド(状態変更を伴う予約実行)が同一メソッドに同居。テストで「確認のつもりが実際にDBを書き換えていた」事故が起きやすい
3null返却(Ch13) — 在庫不足・SKU未存在時に None を返す。呼び出し側が None チェックを怠ると TypeError が発生する
4戻り値スキーマが分岐依存(型安全性なし)dry_run の値によって辞書のキー名が available / reserved と変わり、呼び出し側で分岐ごとに別キーを意識する必要がある
5notify フラグの同居(単一責任違反) — 通知の要否という別の関心事がメソッドに混入している
6失敗理由が区別不可 — SKU未存在と在庫不足という異なる原因が同じ None に潰され、呼び出し側でエラーメッセージを出し分けられない

ヒント(段階的開示)

ヒント1 — 方向性
dry_run のようなブールフラグがメソッドの引数にあり、それによって戻り値の型や副作用の有無が変わっている場合、そのメソッドは2つ以上の責務を抱えている合図。「確認したい」と「実行したい」は呼び出し側の意図として全く別物であり、別メソッド・できれば別クラスに分けるべき。また「値が見つからない/条件を満たさない」ケースで None を返す設計は、呼び出し側に「毎回 is None チェックする」という規律を強制するが、その規律は必ずどこかで破られる。呼び出し不可能な状態は型システムか例外で表現し、「呼べてしまうが失敗する」余地を消す方向で考える。
ヒント2 — アプローチ
  • クエリ側: class StockAvailabilityChecker: def is_available(self, sku: str, qty: int) -> bool: — DBの get のみ呼び、状態変更は一切しない
  • コマンド側: class StockReservationService: def reserve(self, sku: str, qty: int) -> ReservationResult: — 成功時のみ戻り値を返し、失敗時は例外を送出
  • 例外階層: class StockError(Exception)SkuNotFoundError / InsufficientStockError(sku, requested_qty, available_qty)
  • 値オブジェクト: @dataclass(frozen=True, slots=True) class ReservationResult: sku: str; reserved_qty: int; remaining_qty: int
  • Protocol: class StockRepository(Protocol): def get_stock(self, sku: str) -> StockRow | None: ... / def save_stock(self, sku: str, row: StockRow) -> None: ...
  • 通知は reserve() の戻り値を受け取った呼び出し側(または ReservationNotifier.notify(result))が別途呼ぶ
ヒント3 — コードの骨格
class StockError(Exception):
    """在庫関連エラーの基底クラス。"""

class SkuNotFoundError(StockError):
    def __init__(self, sku: str) -> None:
        super().__init__(f"SKU not found: {sku}")
        self.sku = sku

class InsufficientStockError(StockError):
    def __init__(self, sku: str, requested_qty: int, available_qty: int) -> None:
        super().__init__(
            f"Insufficient stock for {sku}: requested={requested_qty}, available={available_qty}"
        )
        self.sku = sku
        self.requested_qty = requested_qty
        self.available_qty = available_qty


@dataclass(frozen=True, slots=True)
class ReservationResult:
    sku: str
    reserved_qty: int
    remaining_qty: int


class StockAvailabilityChecker:
    """クエリ: 在庫確認のみ行い、副作用を一切持たない。"""
    def __init__(self, repository: StockRepository) -> None:
        self._repository = repository

    def is_available(self, sku: str, qty: int) -> bool:
        row = self._repository.get_stock(sku)
        if row is None:
            return False
        return row["quantity"] >= qty


class StockReservationService:
    """コマンド: 在庫予約の確定のみ行い、必ず ReservationResult を返すか例外を送出する。"""
    def __init__(self, repository: StockRepository) -> None:
        self._repository = repository

    def reserve(self, sku: str, qty: int) -> ReservationResult:
        row = self._repository.get_stock(sku)
        if row is None:
            raise SkuNotFoundError(sku)
        if row["quantity"] < qty:
            raise InsufficientStockError(sku, qty, row["quantity"])
        row["quantity"] -= qty
        self._repository.save_stock(sku, row)
        return ReservationResult(sku=sku, reserved_qty=qty, remaining_qty=row["quantity"])

問題点分析(6点)

#問題点分類改善方法
1フラグ引数 dry_run で責務が切り替わるフラグ引数 Ch13別メソッド is_available / reserve に分割
2クエリとコマンドが同一メソッドに同居CQS違反 Ch13別クラス StockAvailabilityChecker / StockReservationService
3失敗時に None を返すnull返却 Ch13専用例外 SkuNotFoundError / InsufficientStockError
4戻り値スキーマが分岐依存(dict のキーが不定)型安全性 Ch4ReservationResult(frozen=True, slots=True)
5notify フラグの同居(単一責任違反)責務分離ReservationNotifier として分離
6失敗理由(SKU未存在 vs 在庫不足)が区別不可例外設計 Ch10InsufficientStockError.available_qty で文脈情報を保持

設計図(SVG)

handle_flash_sale_order 呼び出し側(意図明示) ① 確認 StockAvailabilityChecker クエリ・副作用なし is_available(sku, qty) → bool ② 確定 StockReservationService コマンド・副作用あり reserve(sku, qty) → ReservationResult or raise StockError StockRepository Protocol get_stock / save_stock InMemoryStockRepository テスト用 / 本番は Cloud SQL 等 implements StockError SkuNotFoundError (sku) InsufficientStockError (sku, requested_qty, available_qty) raises ReservationResult frozen=True, slots=True sku / reserved_qty / remaining_qty ReservationNotifier 通知は呼び出し側の任意判断 notify(result) 凡例 実線: 呼び出し / 破線: implements・raises

模範解答

Before — フラグ引数 × CQS違反 × null返却
class InventoryService:
    def __init__(self, db):
        self.db = db

    # 問題1・2: dry_run で「確認」と「予約確定」を切替(フラグ引数・CQS違反)
    # 問題5: notify も同居(責務混在)
    def process_reservation(self, sku, qty, dry_run=False, notify=False):
        row = self.db.get_stock(sku)
        if row is None:              # 問題3・6: null返却、失敗理由が区別不可
            return None
        current_qty = row["quantity"]
        if current_qty < qty:
            return None

        if dry_run:
            # 問題4: dry_run 時とそうでない時でキー名が違う
            return {"sku": sku, "available": True, "requested_qty": qty}

        row["quantity"] = current_qty - qty
        self.db.save_stock(sku, row)
        if notify:
            send_slack_notification(f"{sku} を {qty} 個予約しました")
        return {"sku": sku, "reserved": True, "remaining_qty": row["quantity"]}


def handle_flash_sale_order(service, sku, qty):
    result = service.process_reservation(sku, qty)
    if result["reserved"]:  # result が None なら TypeError
        print(f"予約成功: {sku}")
After — CQS分離 × フラグ引数廃止 × 専用例外階層
"""inventory_reservation.py — フラッシュセール在庫予約(在庫引当)

良いコード・悪いコードで学ぶ設計入門(改訂新版)
  Ch13: メソッド設計 — CQS、フラグ引数の廃止、null返却禁止
  Ch10: 例外の握り潰し回避(専用例外で失敗理由を型として表現)
  Ch4:  frozen dataclass(slots=True) による値オブジェクト
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Final, Protocol, TypedDict

MIN_RESERVATION_QTY: Final[int] = 1

class StockRow(TypedDict):
    sku: str
    quantity: int

# ── Protocol(依存性の注入)────────────────────────────────
class StockRepository(Protocol):
    def get_stock(self, sku: str) -> StockRow | None: ...
    def save_stock(self, sku: str, row: StockRow) -> None: ...

# ── 例外階層(null返却禁止の代替表現・Ch13)─────────────────
class StockError(Exception):
    """在庫関連エラーの基底クラス。"""

class SkuNotFoundError(StockError):
    def __init__(self, sku: str) -> None:
        super().__init__(f"SKU not found: {sku!r}")
        self.sku = sku

class InsufficientStockError(StockError):
    """Attributes:
        available_qty: 実際に確保可能な数量(呼び出し側が UI に表示可能)
    """
    def __init__(self, sku: str, requested_qty: int, available_qty: int) -> None:
        super().__init__(
            f"Insufficient stock for {sku!r}: "
            f"requested={requested_qty}, available={available_qty}"
        )
        self.sku = sku
        self.requested_qty = requested_qty
        self.available_qty = available_qty

# ── 値オブジェクト(Ch4: 不変の活用)───────────────────────
@dataclass(frozen=True, slots=True)
class ReservationResult:
    sku: str
    reserved_qty: int
    remaining_qty: int

# ── クエリ側 — 副作用なし(Ch13: CQS)───────────────────────
class StockAvailabilityChecker:
    def __init__(self, repository: StockRepository) -> None:
        self._repository = repository

    def is_available(self, sku: str, qty: int) -> bool:
        """在庫確認のみ行う(読み取り専用、状態変更なし)。"""
        row = self._repository.get_stock(sku)
        if row is None:
            return False  # クエリなので失敗も bool で表現、例外にしない
        return row["quantity"] >= qty

    def get_available_qty(self, sku: str) -> int:
        row = self._repository.get_stock(sku)
        return row["quantity"] if row is not None else 0

# ── コマンド側 — 副作用あり、null を返さない(Ch13)─────────
class StockReservationService:
    def __init__(self, repository: StockRepository) -> None:
        self._repository = repository

    def reserve(self, sku: str, qty: int) -> ReservationResult:
        """成功時は必ず ReservationResult を返し、失敗時は必ず StockError を送出する。

        Raises:
            ValueError: qty が MIN_RESERVATION_QTY 未満
            SkuNotFoundError: SKU が存在しない
            InsufficientStockError: 在庫不足
        """
        if qty < MIN_RESERVATION_QTY:
            raise ValueError(f"qty must be >= {MIN_RESERVATION_QTY}, got {qty}")

        row = self._repository.get_stock(sku)
        if row is None:
            raise SkuNotFoundError(sku)
        if row["quantity"] < qty:
            raise InsufficientStockError(sku, requested_qty=qty, available_qty=row["quantity"])

        row["quantity"] -= qty
        self._repository.save_stock(sku, row)
        return ReservationResult(sku=sku, reserved_qty=qty, remaining_qty=row["quantity"])

# ── 通知は別責務(単一責任・Ch13)───────────────────────────
class ReservationNotifier:
    def notify(self, result: ReservationResult) -> None:
        print(f"[Slack通知] {result.sku} を {result.reserved_qty} 個予約しました"
              f"(残り {result.remaining_qty} 個)")

# ── 呼び出し側 — 意図が明確 ─────────────────────────────────
def handle_flash_sale_order(checker, reservation_service, notifier, sku, qty) -> None:
    if not checker.is_available(sku, qty):
        print(f"予約不可: {sku} の在庫が不足しています({checker.get_available_qty(sku)} 個のみ)")
        return
    try:
        result = reservation_service.reserve(sku, qty)
    except InsufficientStockError as e:
        print(f"予約失敗(在庫競合): {e}")  # is_available と reserve の間の競合状態
        return
    notifier.notify(result)  # 通知は呼び出し側の任意判断
repository = InMemoryStockRepository(initial={"SKU-001": 5})
checker = StockAvailabilityChecker(repository)
reservation_service = StockReservationService(repository)
notifier = ReservationNotifier()

# ケース1: 在庫確認のみ(副作用なし) → 何度呼んでも在庫は減らない
print(checker.is_available("SKU-001", 3))  # True
print(checker.is_available("SKU-001", 3))  # True(副作用なしなので変化しない)

# ケース2: 予約確定 → 成功
handle_flash_sale_order(checker, reservation_service, notifier, "SKU-001", 3)
# [Slack通知] SKU-001 を 3 個予約しました(残り 2 個)

# ケース3: 在庫不足 → None ではなく False / 例外で表現される
handle_flash_sale_order(checker, reservation_service, notifier, "SKU-001", 10)
# 予約不可: SKU-001 の在庫が不足しています(2 個のみ)

# ケース4: 存在しない SKU → 専用例外で理由が明確
try:
    reservation_service.reserve("SKU-404", 1)
except SkuNotFoundError as e:
    print(f"エラー: {e}")
# エラー: SKU not found: 'SKU-404'

# ケース5: 在庫不足の詳細情報を呼び出し側が取得できる
try:
    reservation_service.reserve("SKU-001", 100)
except InsufficientStockError as e:
    print(f"エラー: {e} / available_qty={e.available_qty}")
# エラー: Insufficient stock for 'SKU-001': requested=100, available=2 / available_qty=2
ポイント適用した設計原則/パターン書籍対応章
is_available / reserve を別クラスに分割コマンド・クエリ分離(CQS)Ch13
dry_run / notify フラグを削除し別メソッド・別クラス化フラグ引数の廃止・単一責任Ch13
reserve() は None を返さず例外を送出null返却禁止Ch13
InsufficientStockError.available_qty例外に文脈情報を持たせるCh10
ReservationResult(frozen=True, slots=True)不変の活用・値オブジェクトCh4
StockRepository(Protocol) 依存性注入密結合解消・テスタビリティCh8系
# tests/test_inventory_reservation.py
import pytest
from inventory_reservation import (
    InMemoryStockRepository, StockAvailabilityChecker, StockReservationService,
    ReservationResult, SkuNotFoundError, InsufficientStockError,
)

class TestStockAvailabilityChecker:
    def setup_method(self):
        self.repo = InMemoryStockRepository(initial={"SKU-001": 5})
        self.checker = StockAvailabilityChecker(self.repo)

    def test_is_available_true(self):
        assert self.checker.is_available("SKU-001", 3) is True

    def test_is_available_false_when_insufficient(self):
        assert self.checker.is_available("SKU-001", 10) is False

    def test_is_available_false_when_sku_missing(self):
        assert self.checker.is_available("SKU-404", 1) is False

    def test_is_available_has_no_side_effect(self):
        # クエリを何度呼んでも在庫は変化しない(CQS 検証)
        self.checker.is_available("SKU-001", 3)
        self.checker.is_available("SKU-001", 3)
        assert self.checker.get_available_qty("SKU-001") == 5


class TestStockReservationService:
    def setup_method(self):
        self.repo = InMemoryStockRepository(initial={"SKU-001": 5})
        self.service = StockReservationService(self.repo)

    def test_reserve_success_returns_result(self):
        result = self.service.reserve("SKU-001", 3)
        assert result == ReservationResult(sku="SKU-001", reserved_qty=3, remaining_qty=2)

    def test_reserve_never_returns_none_on_failure(self):
        with pytest.raises(SkuNotFoundError):
            self.service.reserve("SKU-404", 1)

    def test_reserve_insufficient_stock_raises_with_context(self):
        with pytest.raises(InsufficientStockError) as exc_info:
            self.service.reserve("SKU-001", 100)
        assert exc_info.value.available_qty == 5

    def test_reserve_result_is_frozen(self):
        result = self.service.reserve("SKU-001", 1)
        with pytest.raises(Exception):
            result.reserved_qty = 999  # type: ignore

    def test_reserve_rejects_zero_qty(self):
        with pytest.raises(ValueError):
            self.service.reserve("SKU-001", 0)

ポイント解説

1コマンド・クエリ分離の徹底(Ch13: メソッド設計)is_available(クエリ・副作用なし)と reserve(コマンド・副作用あり)を別クラスに分割。呼び出し側はメソッド名を見ただけで在庫を変更するかどうかを判断できる。
2フラグ引数の廃止(Ch13: メソッド設計)dry_runnotify を削除し、それぞれ別メソッド・別クラス(ReservationNotifier)に分解。フラグ引数は「1メソッドが複数の異なる処理を条件分岐で切り替えている」シグナル。
3null返却禁止・例外による失敗表現(Ch13: メソッド設計)reserve() は成功時は必ず ReservationResult、失敗時は必ず StockError サブクラスを返す契約にし、None チェック漏れによる TypeError を型レベルで防ぐ。is_available() はクエリなので bool のまま(真偽判定自体が正常系)。
4例外に文脈情報を持たせる(Ch10: 例外の握り潰し回避)InsufficientStockError.available_qty により、呼び出し側は「あと何個なら予約できるか」をそのまま UI に反映できる。
5Protocol による依存性注入StockRepository を Protocol 化し、StockAvailabilityChecker / StockReservationService は具体 DB 実装に依存しない。テストは InMemoryStockRepository、本番は Cloud SQL/Spanner 実装に差し替えるだけで済む。
6frozen=True, slots=True の値オブジェクト(Ch4: 不変の活用)ReservationResult を不変にし、予約確定後に呼び出し側が結果を書き換えて後続処理と矛盾させる事故を防ぐ。

実務への応用

フラッシュセールのような高トラフィック・在庫の奪い合いが発生するシーンでは、「在庫確認(表示用)」と「予約確定(決済確定時)」を同じメソッドで扱うと、キャッシュ経由の確認結果と実際の確定処理の整合性検証が難しくなる。CQS で分離しておくことで、is_available は Redis キャッシュ経由の高速パス、reserve は必ず Cloud SQL/Spanner のトランザクションを通す確定パス、というようにインフラ層でも別々の最適化がしやすくなる。

在庫不足時に None ではなく InsufficientStockError(available_qty=...) を送出する設計は、フロントエンドに「あと2個です」というユーザー体験を返す際にも直接活用できる。例外の available_qty をそのまま API レスポンスの detail フィールドにマッピングすればよく、呼び出し側でエラー種別ごとの個別判定ロジックを書かずに済む。

Argo Workflows での再利用: バッチ在庫同期でも StockRepository の Protocol を使えば、本番実行時は BigQuery 実装、CI でのユニットテストは InMemoryStockRepository に差し替えるだけで同じテストコードが再利用できる。

今日のまとめ

フラグ引数でメソッドの挙動を切り替える設計は「呼び出してみるまで副作用が分からない」曖昧さを生む。クエリとコマンドを別メソッド・別クラスに分離し、失敗を None ではなく文脈情報付きの例外で表現することで、呼び出し側の None チェック漏れや責務混在を型と契約のレベルで防げる。

次のステップ

  • 発展問題: 複数SKUを一括予約する reserve_bulk(items: list[tuple[str, int]]) を実装し、途中でエラーが発生した場合に「全て成功 or 全てロールバック」を保証するトランザクション境界の設計(原子性の担保)を考える。asyncio.TaskGroup での並列在庫確認と組み合わせる発展も可能。
  • 参考: 「良いコード・悪いコードで学ぶ設計入門(改訂新版)」Ch4・Ch8・Ch10・Ch13 / Python dataclasses 公式ドキュメント(frozen, slots) / typing.Protocol PEP 544

自己評価(あとで記入)