概要
StrEnum + 遷移テーブルで if-elif 根絶(Ch6)
if reservation["status"] == "reserved": ... の文字列比較と cancel()/fulfill() での重複分岐を、VALID_TRANSITIONS 辞書と共通 _transition() ヘルパーで1箇所に集約する。
frozen dataclass + replace() でイミュータブル更新(Ch10)
ステータス変更は r["status"] = "cancelled" の直接変更ではなく replace(r, status=...) で新インスタンスを生成。副作用を局所化し不変性を保証する。
Counter でループ集計を1行に圧縮(Ch7)
手動の if sku in result: result[sku] += qty else: result[sku] = qty を Counter({r.sku_id: r.qty for r in ... if r.status == RESERVED}) の1表現に置き換える。
datetime.UTC で aware datetime に統一(Ch10)
datetime.now() の naive datetime はタイムゾーン比較・保存時にバグを引き起こす。datetime.now(tz=datetime.UTC) で UTC aware に統一しDST問題を防ぐ。
問題
ECサイトの在庫管理システムに存在する ReservationManager クラスは、商品の予約・引き当て・キャンセルを管理するが複数の設計上の問題を抱えている。問題点を全て洗い出し、Ch6(条件分岐の整理)・Ch7(コレクション操作の整理)・Ch10(不変と副作用) を適用してリファクタリングせよ。
制約・前提条件
- Python 3.12+、型ヒント・
dataclassを使うこと StrEnumでステータスを型安全に管理することdataclass(frozen=True, slots=True)で予約レコードを値オブジェクト化することdatetime.now(tz=datetime.UTC)で aware datetime に統一することCounterまたはdefaultdictで集計ループを整理すること- 遷移テーブルで条件分岐を1箇所に集約すること
- Google スタイル docstring・インラインコメント・名前付き定数を含めること
期待する回答形式: 問題点の列挙(番号付き)+ 改善後コード(Google スタイル docstring・インラインコメント・名前付き定数含む)+ 実行例(input→output)+ 適用した設計パターン名と書籍の対応章
悪いコード (Before)
このコードには 6つの設計上の問題 が隠れています。見つけてみてください。
bad_reservation_manager.py — 問題だらけの予約管理
import datetime
class ReservationManager:
def __init__(self):
self.reservations = {} # order_id -> dict
self.stock = {} # sku_id -> int
def reserve(self, order_id, sku_id, qty):
# 問題①: 引数バリデーションなし
if sku_id in self.stock:
if self.stock[sku_id] >= qty:
self.stock[sku_id] = self.stock[sku_id] - qty
self.reservations[order_id] = {
"sku_id": sku_id,
"qty": qty,
"status": "reserved",
"created_at": datetime.datetime.now() # 問題②: naive datetime
}
return True
else:
return False
else:
return False
def cancel(self, order_id):
# 問題③: 直接アクセス → KeyError の危険
reservation = self.reservations[order_id]
# 問題④: マジックストリング "reserved"/"cancelled"/"fulfilled"
if reservation["status"] == "reserved":
self.stock[reservation["sku_id"]] += reservation["qty"]
reservation["status"] = "cancelled"
return True
elif reservation["status"] == "cancelled":
return False
elif reservation["status"] == "fulfilled":
return False
else:
return False
def get_summary(self):
# 問題⑤: 手動集計ループ(Counter/defaultdict 未使用)
result = {}
for order_id in self.reservations:
r = self.reservations[order_id]
sku = r["sku_id"]
if sku in result:
result[sku] = result[sku] + r["qty"]
else:
result[sku] = r["qty"]
return result
def fulfill(self, order_id):
# 問題⑥: cancel() と同じ if-elif が重複(DRY違反)
r = self.reservations[order_id]
if r["status"] == "reserved":
r["status"] = "fulfilled"
return True
elif r["status"] == "fulfilled":
return False
elif r["status"] == "cancelled":
return False
else:
return False
問題点サマリー(6点)
1引数バリデーションが皆無 —
qty=-1 や空文字 order_id を渡しても何も検出されない。ValueError / OutOfStockError を明示的に送出すること2
datetime.now()(naive datetime) — タイムゾーン情報なし。本番環境での DST や UTC 比較で問題発生。datetime.now(tz=datetime.UTC) に統一3
self.reservations[order_id] 直接アクセス — 存在しない order_id で KeyError。.get() でカスタム例外に変換すること4マジックストリング
"reserved"/"cancelled"/"fulfilled" — typo が実行時まで検出不可。StrEnum ReservationStatus に置き換える5手動集計ループ(if-else でキー存在チェック) —
Counter で1表現に圧縮できる(Ch7)6
cancel() と fulfill() に同じ if-elif が重複 — DRY 違反。遷移テーブル + 共通 _transition() に集約すること(Ch6)ヒント(段階的開示)
ヒント1 — 方向性
ステータス管理は
str のマジックストリングではなく StrEnum で型安全にできる。cancel と fulfill に同じ条件分岐が2回出てくるのは、ステータス遷移ロジックが分散しているサイン。naive datetime は本番環境でタイムゾーン問題を引き起こす。dataclass(frozen=True) を使うとステータスの直接書き換えができなくなるが、replace() でイミュータブル更新ができる。
ヒント2 — アプローチ
StrEnumでReservationStatusを定義し、比較に文字列リテラルを使わなくする- 遷移許可テーブル
VALID_TRANSITIONS: dict[ReservationStatus, frozenset[ReservationStatus]]でステータス遷移を一元管理 @dataclass(frozen=True, slots=True)でReservationを値オブジェクト化。更新はreplace(r, status=new_status)で行うdatetime.now(tz=datetime.UTC)で aware datetime に統一Counter({r.sku_id: r.qty for r in ... if r.status == RESERVED})で集計を1行に圧縮
ヒント3 — コードの骨格
from enum import StrEnum
from dataclasses import dataclass, replace
import datetime
from collections import Counter
from typing import Final
class ReservationStatus(StrEnum):
RESERVED = "reserved"
FULFILLED = "fulfilled"
CANCELLED = "cancelled"
VALID_TRANSITIONS: Final[dict[ReservationStatus, frozenset[ReservationStatus]]] = {
ReservationStatus.RESERVED: frozenset({
ReservationStatus.FULFILLED, ReservationStatus.CANCELLED
}),
ReservationStatus.FULFILLED: frozenset(),
ReservationStatus.CANCELLED: frozenset(),
}
@dataclass(frozen=True, slots=True)
class Reservation:
order_id: str
sku_id: str
qty: int
status: ReservationStatus
created_at: datetime.datetime # aware datetime (UTC)
class ReservationManager:
def _transition(self, order_id: str, new_status: ReservationStatus) -> Reservation:
r = self._reservations.get(order_id)
if r is None:
raise ReservationNotFoundError(...)
if new_status not in VALID_TRANSITIONS[r.status]:
raise InvalidTransitionError(...)
updated = replace(r, status=new_status)
self._reservations[order_id] = updated
return updated
問題点分析(6点)
| # | 問題点 | 分類 | 改善方法 |
|---|---|---|---|
| 1 | 引数バリデーションなし(qty=-1、空 order_id 等を検出できない) | 堅牢性 | ValueError / OutOfStockError を明示送出 |
| 2 | datetime.now() の naive datetime | Ch10 不変/副作用 | datetime.now(tz=datetime.UTC) で UTC aware に統一 |
| 3 | self.reservations[order_id] 直接アクセス → KeyError | Ch7 コレクション | .get() + ReservationNotFoundError に変換 |
| 4 | マジックストリング "reserved" / "cancelled" / "fulfilled" | Ch6 条件分岐 | StrEnum ReservationStatus で型安全に管理 |
| 5 | 手動集計ループ(if-else でキー存在チェック) | Ch7 コレクション | Counter で1表現に圧縮 |
| 6 | cancel() と fulfill() に同じ if-elif が重複(DRY 違反) | Ch6 条件分岐 | 遷移テーブル + _transition() に集約 |
模範解答
"""reservation_manager.py — 在庫予約管理(Bad → Good リファクタリング)
良いコード・悪いコードで学ぶ設計入門(改訂新版)
- Ch6: 条件分岐の整理(遷移テーブル × StrEnum で if-elif 根絶)
- Ch7: コレクション操作の整理(Counter × dict.get() × プライベート化)
- Ch10: 不変と副作用(frozen dataclass × aware datetime × replace())
"""
from __future__ import annotations
import datetime
from collections import Counter
from dataclasses import dataclass, replace
from enum import StrEnum
from typing import Final
# ── 名前付き定数 ────────────────────────────────────────────────────────────
DEFAULT_STOCK: Final[int] = 0 # 未登録 SKU の初期在庫
# ── StrEnum でステータスを型安全に管理(Ch6: 文字列リテラル比較を廃絶)──────
class ReservationStatus(StrEnum):
RESERVED = "reserved" # 在庫引き当て済み・未確定
FULFILLED = "fulfilled" # 出荷確定済み
CANCELLED = "cancelled" # キャンセル済み(在庫返却済み)
# ── 遷移テーブルで条件分岐を1箇所に集約(Ch6: DRY 違反を解消)──────────────
VALID_TRANSITIONS: Final[dict[ReservationStatus, frozenset[ReservationStatus]]] = {
ReservationStatus.RESERVED: frozenset({
ReservationStatus.FULFILLED,
ReservationStatus.CANCELLED,
}),
ReservationStatus.FULFILLED: frozenset(), # 終端状態
ReservationStatus.CANCELLED: frozenset(), # 終端状態
}
# ── 値オブジェクト(Ch10: frozen=True で不変性を保証)──────────────────────
@dataclass(frozen=True, slots=True)
class Reservation:
"""予約1件を表す不変の値オブジェクト。
Attributes:
order_id: 注文ID(一意キー)。
sku_id: 在庫管理単位ID。
qty: 引き当て数量。
status: 現在の予約ステータス。
created_at: 予約作成日時(aware datetime, UTC)。
"""
order_id: str
sku_id: str
qty: int
status: ReservationStatus
created_at: datetime.datetime # 必ず UTC aware datetime
# ── カスタム例外 ──────────────────────────────────────────────────────────
class OutOfStockError(Exception):
"""在庫不足で予約できない場合に送出する。"""
class ReservationNotFoundError(KeyError):
"""指定された order_id の予約が存在しない場合に送出する。"""
class InvalidTransitionError(ValueError):
"""許可されていないステータス遷移を試みた場合に送出する。"""
# ── マネージャ本体 ──────────────────────────────────────────────────────────
class ReservationManager:
"""在庫の予約・引き当て・キャンセルを管理するクラス。"""
def __init__(self, initial_stock: dict[str, int] | None = None) -> None:
# プライベート化でカプセル化(Ch10)
self._stock: dict[str, int] = dict(initial_stock or {})
self._reservations: dict[str, Reservation] = {}
def _transition(
self,
order_id: str,
new_status: ReservationStatus,
) -> Reservation:
"""予約のステータスを遷移させる共通ロジック(Ch6: 条件分岐の集約)。
cancel() と fulfill() の重複 if-elif をここに集約する。
"""
r = self._reservations.get(order_id) # dict.get() で安全アクセス(Ch7)
if r is None:
raise ReservationNotFoundError(f"予約が見つかりません: {order_id!r}")
# 遷移テーブルを参照して許可チェック(if-elif は1行にもならない)
if new_status not in VALID_TRANSITIONS[r.status]:
raise InvalidTransitionError(
f"{r.status!r} → {new_status!r} は許可されていない遷移です"
)
# frozen dataclass は replace() でイミュータブルに更新する(Ch10)
updated = replace(r, status=new_status)
self._reservations[order_id] = updated
return updated
def add_stock(self, sku_id: str, qty: int) -> None:
"""在庫を追加登録する。"""
self._stock[sku_id] = self._stock.get(sku_id, DEFAULT_STOCK) + qty
def reserve(self, order_id: str, sku_id: str, qty: int) -> Reservation:
"""在庫を引き当てて予約を作成する。"""
# ── 引数バリデーション(問題①を解消)──────────────────────────────
if not order_id or not sku_id:
raise ValueError("order_id と sku_id は空文字列にできません")
if qty <= 0:
raise ValueError(f"qty は正の整数である必要があります: {qty}")
if order_id in self._reservations:
raise ValueError(f"order_id は既に予約済みです: {order_id!r}")
available = self._stock.get(sku_id, DEFAULT_STOCK)
if available < qty:
raise OutOfStockError(
f"在庫不足: sku_id={sku_id!r}, 要求={qty}, 在庫={available}"
)
self._stock[sku_id] = available - qty
reservation = Reservation(
order_id=order_id,
sku_id=sku_id,
qty=qty,
status=ReservationStatus.RESERVED,
# aware datetime で naive datetime 問題②を解消
created_at=datetime.datetime.now(tz=datetime.UTC),
)
self._reservations[order_id] = reservation
return reservation
def cancel(self, order_id: str) -> Reservation:
"""予約をキャンセルし在庫を戻す。"""
# 共通遷移ロジックに委譲(問題③④⑥を解消)
updated = self._transition(order_id, ReservationStatus.CANCELLED)
# キャンセル時のみ在庫を戻す
self._stock[updated.sku_id] = (
self._stock.get(updated.sku_id, DEFAULT_STOCK) + updated.qty
)
return updated
def fulfill(self, order_id: str) -> Reservation:
"""予約を出荷確定に遷移させる。"""
# cancel() との重複が完全に解消される(問題⑥)
return self._transition(order_id, ReservationStatus.FULFILLED)
def get_summary(self) -> dict[str, int]:
"""SKU 別の予約中(RESERVED)数量合計を返す。"""
# Counter で手動ループを1表現に圧縮(問題⑤を解消: Ch7)
return dict(
Counter(
{r.sku_id: r.qty
for r in self._reservations.values()
if r.status == ReservationStatus.RESERVED}
)
)
正常系
mgr = ReservationManager(initial_stock={"SKU-001": 10, "SKU-002": 5})
r1 = mgr.reserve("ORD-001", "SKU-001", 3)
r2 = mgr.reserve("ORD-002", "SKU-001", 2)
r3 = mgr.reserve("ORD-003", "SKU-002", 5)
print(mgr.get_summary())
# {'SKU-001': 5, 'SKU-002': 5}
mgr.fulfill("ORD-001")
print(mgr.get_summary())
# {'SKU-001': 2, 'SKU-002': 5} ← fulfilled は除外
mgr.cancel("ORD-003")
print(mgr._stock)
# {'SKU-001': 5, 'SKU-002': 5} ← SKU-002 の在庫が戻る
異常系
# 在庫不足
try:
mgr.reserve("ORD-004", "SKU-001", 999)
except OutOfStockError as e:
print(e)
# 在庫不足: sku_id='SKU-001', 要求=999, 在庫=5
# 不正な遷移(fulfilled → cancel)
try:
mgr.cancel("ORD-001")
except InvalidTransitionError as e:
print(e)
# 'fulfilled' → 'cancelled' は許可されていない遷移です
# naive vs aware datetime の違い
print(r1.created_at.tzinfo) # UTC(タイムゾーン情報あり)
naive = datetime.datetime.now()
print(naive.tzinfo) # None(問題があるパターン)
| ポイント | 適用した設計原則/パターン | 書籍対応章 |
|---|---|---|
StrEnum ReservationStatus | 型安全なステータス管理(文字列リテラル廃絶) | Ch6 条件分岐 |
VALID_TRANSITIONS + _transition() | 状態遷移テーブル + 条件分岐の集約(DRY) | Ch6 条件分岐 |
dict.get() + カスタム例外 | コレクション操作の安全化 | Ch7 コレクション |
Counter で集計1行化 | コレクション操作の整理 | Ch7 コレクション |
Reservation(frozen=True, slots=True) | 値オブジェクト(Value Object) | Ch10 不変/副作用 |
replace(r, status=...) | イミュータブル更新パターン | Ch10 不変/副作用 |
datetime.now(tz=datetime.UTC) | aware datetime(副作用の排除) | Ch10 不変/副作用 |
設計図 — Bad vs Good の変換フロー
ポイント解説
1
StrEnum でステータスを型安全に管理(Ch6)reservation["status"] == "reserved" という文字列比較は "reseved" のような typo が実行時まで検出されない。StrEnum で ReservationStatus を定義すると mypy が静的に検出できる。StrEnum は str のサブクラスなので json.dumps() にもそのまま渡せる点が IntEnum との違い。
2
遷移テーブル +
_transition() で条件分岐を集約(Ch6)cancel() と fulfill() に全く同じ if-elif が重複していた(DRY 違反)。VALID_TRANSITIONS: dict[ReservationStatus, frozenset[ReservationStatus]] に「どのステータスからどのステータスに遷移できるか」を定義し、_transition() ヘルパーで参照することで、新しいステータスを追加するときも辞書の1エントリ追加で済む(OCP の部分的な達成)。
3
frozen=True dataclass + replace() でイミュータブル更新(Ch10)reservation["status"] = "cancelled" の直接変更は副作用が大きく、どこで状態が変わったか追跡しにくい。Reservation(frozen=True) にするとフィールドへの直接代入を禁止できる。更新は dataclasses.replace(r, status=new_status) で新インスタンスを生成し、辞書に再代入する形にすることで変更箇所が明確になる。
4
datetime.now(tz=datetime.UTC) で aware datetime(Ch10)datetime.now() は naive datetime(tzinfo=None)を返す。DB や API が UTC 前提の場合、JSTのサーバーで生成した naive datetime を保存すると9時間ずれる。datetime.now(tz=datetime.UTC)(Python 3.11+の datetime.UTC)で常に UTC aware を生成することで比較・保存時の問題を防ぐ。
5
Counter で集計ループを1行に圧縮(Ch7)if sku in result: result[sku] += qty else: result[sku] = qty という手動集計は Counter + dict 内包表記で Counter({r.sku_id: r.qty for r in ... if r.status == RESERVED}) の1表現に圧縮できる。フィルタ条件(RESERVED のみ)も内包表記に含められるためコードの意図が明確になる。
6
dict.get() + カスタム例外でエラーを明示化(Ch7)self.reservations[order_id] の直接アクセスは存在しない key で KeyError(汎用例外)が発生する。self._reservations.get(order_id) で None を受け取り、ReservationNotFoundError を明示的に送出することで呼び出し元が意図を理解しやすくなる。カスタム例外は KeyError のサブクラスにしておくと既存の except KeyError も捕捉できる。
実務への応用
- Argo Workflows のステータス管理:
_transition()+ 遷移テーブルパターンは Argo Workflows のジョブステータス(Pending→Running→Succeeded/Failed)管理にそのまま応用できる。無効な遷移(Succeeded → Running 等)を実行前に検出できる。 - BigQuery/dbt の在庫集計:
get_summary()のCounterパターンはSELECT sku_id, SUM(qty) FROM reservations WHERE status = 'reserved' GROUP BY sku_idに対応する。Python バッチと BQ クエリで同じ集計ロジックを持つことで dbt Unit Tests との差分テストが書きやすい。 - DataDog カスタムメトリクス:
Reservation.created_at(UTC aware)を使い、予約から出荷確定までのリードタイム(fulfilled_at - created_at)を DataDog に送信できる。naive datetime ではタイムゾーン変換エラーが発生するため aware datetime の徹底が可観測性の前提。 - MOps/販促システムへの応用: クーポン配布ステータス(PENDING→DISTRIBUTED→USED/EXPIRED)も同じ遷移テーブルパターンで管理できる。VALID_TRANSITIONS を設定ファイル(JSON/YAML)から読み込む設計にすれば、コード変更なしに遷移ルールを変更できる。
今日のまとめ
StrEnum × 遷移テーブル VALID_TRANSITIONS で条件分岐の重複を根絶し、frozen=True dataclass × replace() でイミュータブル更新を実現することで、状態管理コードは「型安全・1箇所に集約・副作用なし」の3条件を満たせる。datetime.UTC(aware datetime)と Counter は即座に使える実務パターンとして必ず押さえておくこと。遷移テーブルパターンは在庫管理・ワークフロー管理・クーポン管理など、状態を持つあらゆるドメインで再利用できる汎用的な設計。