概要
コマンド・クエリ分離(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アクセスは
ProtocolでStockRepositoryインターフェースを定義し、依存性注入でテスト容易性を確保すること - Google スタイル docstring・インラインコメント・名前付き定数を含めること
悪いコード (Before)
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}")
None を返す。呼び出し側が None チェックを怠ると TypeError が発生するdry_run の値によって辞書のキー名が available / reserved と変わり、呼び出し側で分岐ごとに別キーを意識する必要がある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 のキーが不定) | 型安全性 Ch4 | ReservationResult(frozen=True, slots=True) |
| 5 | notify フラグの同居(単一責任違反) | 責務分離 | ReservationNotifier として分離 |
| 6 | 失敗理由(SKU未存在 vs 在庫不足)が区別不可 | 例外設計 Ch10 | InsufficientStockError.available_qty で文脈情報を保持 |
設計図(SVG)
模範解答
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}")
"""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)
ポイント解説
is_available(クエリ・副作用なし)と reserve(コマンド・副作用あり)を別クラスに分割。呼び出し側はメソッド名を見ただけで在庫を変更するかどうかを判断できる。dry_run と notify を削除し、それぞれ別メソッド・別クラス(ReservationNotifier)に分解。フラグ引数は「1メソッドが複数の異なる処理を条件分岐で切り替えている」シグナル。reserve() は成功時は必ず ReservationResult、失敗時は必ず StockError サブクラスを返す契約にし、None チェック漏れによる TypeError を型レベルで防ぐ。is_available() はクエリなので bool のまま(真偽判定自体が正常系)。InsufficientStockError.available_qty により、呼び出し側は「あと何個なら予約できるか」をそのまま UI に反映できる。StockRepository を Protocol 化し、StockAvailabilityChecker / StockReservationService は具体 DB 実装に依存しない。テストは InMemoryStockRepository、本番は Cloud SQL/Spanner 実装に差し替えるだけで済む。ReservationResult を不変にし、予約確定後に呼び出し側が結果を書き換えて後続処理と矛盾させる事故を防ぐ。実務への応用
フラッシュセールのような高トラフィック・在庫の奪い合いが発生するシーンでは、「在庫確認(表示用)」と「予約確定(決済確定時)」を同じメソッドで扱うと、キャッシュ経由の確認結果と実際の確定処理の整合性検証が難しくなる。CQS で分離しておくことで、is_available は Redis キャッシュ経由の高速パス、reserve は必ず Cloud SQL/Spanner のトランザクションを通す確定パス、というようにインフラ層でも別々の最適化がしやすくなる。
在庫不足時に None ではなく InsufficientStockError(available_qty=...) を送出する設計は、フロントエンドに「あと2個です」というユーザー体験を返す際にも直接活用できる。例外の available_qty をそのまま API レスポンスの detail フィールドにマッピングすればよく、呼び出し側でエラー種別ごとの個別判定ロジックを書かずに済む。
StockRepository の Protocol を使えば、本番実行時は BigQuery 実装、CI でのユニットテストは InMemoryStockRepository に差し替えるだけで同じテストコードが再利用できる。
今日のまとめ
次のステップ
- 発展問題: 複数SKUを一括予約する
reserve_bulk(items: list[tuple[str, int]])を実装し、途中でエラーが発生した場合に「全て成功 or 全てロールバック」を保証するトランザクション境界の設計(原子性の担保)を考える。asyncio.TaskGroupでの並列在庫確認と組み合わせる発展も可能。 - 参考: 「良いコード・悪いコードで学ぶ設計入門(改訂新版)」Ch4・Ch8・Ch10・Ch13 / Python
dataclasses公式ドキュメント(frozen,slots) /typing.ProtocolPEP 544