概要
defaultdict でカウンタを整理(Ch4)
if cat in result: ... else: ... という存在チェックを defaultdict(Decimal) に置き換え、totals[cat] += subtotal の1行に圧縮する。
sorted + スライスで整理(Ch4)
手動 sort() → reverse() と while カウンタループを sorted(..., reverse=True)[:n] の1表現に置き換える。境界チェックも自動。
frozen dataclass で値オブジェクト化(Ch3)
dict のままでは item["categorry"] のタイポが実行時まで分からない。OrderItem dataclass にすると mypy が静的検出できる。
Decimal で金額精度を保証(Ch2)
float 演算の誤差(例: 2500 * 3 = 7499.99...)を防ぐため Decimal(str(price)) で精度を保持する。
問題
ECサイトの注文集計バッチで、以下の「悪いコード」は商品カテゴリ別の売上集計と上位N件の取得を実装したもの。問題点を全て洗い出し、Ch4(コレクション操作の整理) および Ch2(型の活用) を適用してリファクタリングせよ。
制約・前提条件
- Python 3.12+、型ヒント・
dataclassを使うこと collections.defaultdictを適切に使うこと- 入力データを
dataclass(frozen=True, slots=True)の値オブジェクトとして型安全に扱うこと - Google スタイル docstring・インラインコメント・名前付き定数を含めること
categoryがNoneの場合のデフォルト値を名前付き定数で管理すること
期待する回答形式: 問題点の列挙(番号付き)+ 改善後コード(Google スタイル docstring・インラインコメント・名前付き定数含む)+ 実行例(input→output)+ 適用した設計パターン名と書籍の対応章
悪いコード (Before)
このコードには 6つの設計上の問題 が隠れています。見つけてみてください。
bad_sales_aggregator.py — 問題だらけの売上集計
def get_top_items(orders, n):
result = {}
for o in orders:
for item in o["items"]:
cat = item.get("category")
# 問題①: None チェックに == を使う(is None が正しい)
if cat == None:
cat = "unknown" # 問題⑤: マジックナンバー文字列
# 問題②: if-else でキー存在確認(defaultdict で不要)
if cat in result:
result[cat] = result[cat] + item["price"] * item["qty"]
else:
result[cat] = item["price"] * item["qty"]
# 問題③: sort() + reverse() の2ステップ(sorted+reverse=Trueに統合できる)
sorted_result = []
for k in result:
sorted_result.append((k, result[k]))
sorted_result.sort(key=lambda x: x[1])
sorted_result.reverse()
# 問題④: while カウンタループ(スライス [:n] で十分)
top = []
i = 0
while i < n and i < len(sorted_result):
top.append(sorted_result[i])
i += 1
return top
# 問題⑥: 型ヒントなし・入力が dict のまま(タイポが実行時まで検出不可)
# 問題⑦: float で金額計算(浮動小数点誤差が発生する可能性)
問題点サマリー(6点)
1
cat == None(is None が正しい) — == で None 比較は非推奨。is None または or DEFAULT_CATEGORY が Python イディオム2if-else でキー存在チェック —
defaultdict(Decimal) を使えばキーなしの場合の初期化が不要になる3手動 sort + reverse の2ステップ —
sorted(..., reverse=True) で1行に統合できる4while カウンタループ —
[:n] スライスで置き換えられる(境界チェック自動)5マジックストリング
"unknown" — 名前付き定数 DEFAULT_CATEGORY にするべき6型ヒントなし・dict のまま扱う —
OrderItem dataclass にして静的解析できるようにすることヒント(段階的開示)
ヒント1 — 方向性
カテゴリ別の売上合計を計算するにはマニュアルな辞書操作より
collections.defaultdict(Decimal) を使うとシンプルに書ける。None チェックを毎回 if cat == None と書くより or / is None のイディオムを使うほうがPythonicで誤りが少ない。入力データの各フィールドは型ヒントのない dict で触っていると KeyError が実行時まで分からない。
ヒント2 — アプローチ
item.get("category") or DEFAULT_CATEGORYの1行でNone/ 空文字列を一括処理collections.defaultdict(Decimal)でキー存在チェックを省略- 手動ソート+リバース →
sorted(..., reverse=True)の1行に置き換える while i < nのカウンタループ →[:n]スライスに置き換える- 入力
dictをOrderItem/Orderdataclass に変換して型安全に扱う
ヒント3 — コードの骨格
from collections import defaultdict
from dataclasses import dataclass
from decimal import Decimal
from typing import Final
DEFAULT_CATEGORY: Final[str] = "unknown"
TOP_N_DEFAULT: Final[int] = 10
@dataclass(frozen=True, slots=True)
class OrderItem:
name: str
category: str
price: Decimal
qty: int
@property
def subtotal(self) -> Decimal:
return self.price * self.qty
def aggregate_by_category(items: list[OrderItem]) -> dict[str, Decimal]:
totals: defaultdict[str, Decimal] = defaultdict(Decimal)
for item in items:
totals[item.category] += item.subtotal
return dict(totals)
def get_top_categories(orders: list[Order], n: int = TOP_N_DEFAULT):
totals = aggregate_by_category(...)
return sorted(totals.items(), key=lambda kv: kv[1], reverse=True)[:n]
問題点分析(6点)
| # | 問題点 | 分類 | 改善方法 |
|---|---|---|---|
| 1 | cat == None(is None が正しい Python イディオム) | Pythonic 書き方 | item.get("category") or DEFAULT_CATEGORY に統合 |
| 2 | if-else でキー存在確認 | コレクション Ch4 | defaultdict(Decimal) で自動初期化 |
| 3 | sort() + reverse() の2ステップ | コレクション Ch4 | sorted(..., reverse=True) の1行に統合 |
| 4 | while カウンタループ | コレクション Ch4 | [:n] スライスで置き換え |
| 5 | マジックストリング "unknown" | 可読性 Ch2 | DEFAULT_CATEGORY: Final[str] 定数に名前付け |
| 6 | 型ヒントなし・dict のまま扱う | 型安全性 Ch2/Ch3 | OrderItem dataclass (frozen+slots) で値オブジェクト化 |
模範解答
"""sales_aggregator.py — カテゴリ別売上集計(Bad → Good リファクタリング)
良いコード・悪いコードで学ぶ設計入門(改訂新版)
- Ch2: 型の活用(dataclass frozen+slots, Decimal, type hints)
- Ch4: コレクション操作の整理(defaultdict, sorted, スライス)
- Ch3: 値オブジェクト(OrderItem を immutable dataclass で表現)
"""
from __future__ import annotations
from collections import defaultdict
from dataclasses import dataclass
from decimal import Decimal
from typing import Final
# ── 名前付き定数(マジックナンバー・魔法の文字列を廃絶)────────────────────
DEFAULT_CATEGORY: Final[str] = "unknown" # category が None/未設定の場合に使うラベル
TOP_N_DEFAULT: Final[int] = 10 # 上位件数のデフォルト値
# ── 値オブジェクト(Ch3: 不変・型安全なデータ表現)──────────────────────────
@dataclass(frozen=True, slots=True)
class OrderItem:
"""注文内の1品目を表す不変の値オブジェクト。
Attributes:
name: 商品名。
category: 商品カテゴリ(None の場合は DEFAULT_CATEGORY に正規化して渡す)。
price: 単価(Decimal で金額精度を保証)。
qty: 数量。
"""
name: str
category: str
price: Decimal
qty: int
@property
def subtotal(self) -> Decimal:
"""小計(単価 × 数量)を返す。Decimal で精度を保持。"""
return self.price * self.qty
@dataclass(frozen=True, slots=True)
class Order:
"""1注文を表す不変の値オブジェクト。
Attributes:
order_id: 注文ID。
items: 注文品目リスト(tuple で immutable 化)。
"""
order_id: str
items: tuple[OrderItem, ...]
# ── パース関数(dict → 値オブジェクトへの変換)──────────────────────────────
def parse_order(raw: dict) -> Order:
"""生の dict を Order 値オブジェクトに変換する。
Args:
raw: API / DB から取得した注文 dict。
Returns:
型安全な Order 値オブジェクト。
Raises:
KeyError: 必須フィールドが欠けている場合。
"""
items = tuple(
OrderItem(
name=item["name"],
# None / 空文字列を DEFAULT_CATEGORY に正規化(or で一括処理)
category=item.get("category") or DEFAULT_CATEGORY,
price=Decimal(str(item["price"])), # float → str → Decimal で精度を保持
qty=int(item["qty"]),
)
for item in raw["items"]
)
return Order(order_id=raw["id"], items=items)
# ── 集計関数(Ch4: defaultdict でカウンタ操作を整理)────────────────────────
def aggregate_by_category(orders: list[Order]) -> dict[str, Decimal]:
"""全注文のカテゴリ別売上合計を集計する。
Args:
orders: 集計対象の注文リスト。
Returns:
カテゴリ名 → 売上合計(Decimal)のマッピング。
"""
# defaultdict(Decimal) でキー存在チェックを省略
# if cat in result: ... else: result[cat] = ... の分岐がゼロになる
totals: defaultdict[str, Decimal] = defaultdict(Decimal)
for order in orders:
for item in order.items:
totals[item.category] += item.subtotal # 初回アクセス時に Decimal('0') が自動挿入
return dict(totals)
def get_top_categories(
orders: list[Order], n: int = TOP_N_DEFAULT
) -> list[tuple[str, Decimal]]:
"""売上上位 n カテゴリを降順で返す。
Args:
orders: 集計対象の注文リスト。
n: 返す上位件数(デフォルト: TOP_N_DEFAULT = 10)。
Returns:
(カテゴリ名, 売上合計) のタプルリスト(降順、最大 n 件)。
"""
totals = aggregate_by_category(orders)
# sorted + reverse=True の1行で手動sort+reverseを置き換え(Ch4)
# [:n] スライスで while カウンタループを置き換え(境界チェック自動)
return sorted(totals.items(), key=lambda kv: kv[1], reverse=True)[:n]
# ── 実行例 ─────────────────────────────────────────────────────────────────
if __name__ == "__main__":
raw_orders = [
{"id": "ORD-001", "items": [
{"name": "T-shirt", "category": "apparel", "price": 2500, "qty": 3},
{"name": "Jeans", "category": None, "price": 5800, "qty": 1},
]},
{"id": "ORD-002", "items": [
{"name": "Sneakers", "category": "apparel", "price": 12000, "qty": 2},
{"name": "Book", "category": "books", "price": 1500, "qty": 5},
{"name": "Gadget", "category": "electronics", "price": 8000, "qty": 1},
]},
]
orders = [parse_order(r) for r in raw_orders]
results = get_top_categories(orders, n=3)
for rank, (cat, total) in enumerate(results, start=1):
print(f"{rank}. {cat}: ¥{total:,}")
正常系(集計 + ランキング)
orders = [parse_order(r) for r in raw_orders]
results = get_top_categories(orders, n=3)
for rank, (cat, total) in enumerate(results, start=1):
print(f"{rank}. {cat}: ¥{total:,}")
出力
1. apparel: ¥31,500 # T-shirt(2500×3) + Sneakers(12000×2) = 7500 + 24000
2. unknown: ¥5,800 # Jeans(5800×1) — category=None → DEFAULT_CATEGORY
3. books: ¥7,500 # Book(1500×5)
# ※ electronics(8000) は rank=4 なので n=3 では除外される
None カテゴリの正規化確認
# Bad: category=None の Jeans が "unknown" に正規化されているか確認
item = parse_order({"id": "ORD-001", "items": [
{"name": "Jeans", "category": None, "price": 5800, "qty": 1}
]}).items[0]
print(item.category) # "unknown"(DEFAULT_CATEGORY に正規化済み)
print(item.subtotal) # Decimal('5800')(float 誤差なし)
Bad: float 誤差の例
# Bad(float)
print(2500.0 * 3 * 0.1) # 749.9999999999999 ← 誤差!
# Good(Decimal)
from decimal import Decimal
print(Decimal("2500") * 3 * Decimal("0.1")) # 750.0 ← 正確
| ポイント | 適用した設計原則/パターン | 書籍対応章 |
|---|---|---|
defaultdict(Decimal) で if-else 省略 | コレクション操作の整理 | Ch4 コレクション |
sorted(..., reverse=True)[:n] | コレクション操作の整理 | Ch4 コレクション |
or DEFAULT_CATEGORY で None 正規化 | 型の活用(名前付き定数) | Ch2 型の活用 |
Decimal(str(price)) で金額精度保証 | 型の活用(精度保証型) | Ch2 型の活用 |
OrderItem(frozen=True, slots=True) | 値オブジェクト(Value Object) | Ch3 値オブジェクト |
subtotal プロパティに計算を局所化 | 計算ロジックの局所化(SRP) | Ch3/Ch6 |
設計図 — Bad vs Good の変換フロー
ポイント解説
1
defaultdict(Decimal) でキー存在チェックを省略(Ch4)if cat in result: result[cat] += ... else: result[cat] = ... というパターンは defaultdict を使うと totals[cat] += item.subtotal の1行に圧縮できる。初回アクセス時に Decimal('0') が自動挿入されるため分岐が不要。辞書操作の分岐が減るほどコードが短くなりバグ混入箇所も減る。
2
sorted(..., reverse=True)[:n] で手動ソート+ループを置き換え(Ch4)sort() → reverse() の2行と while i < n: top.append(...); i += 1 のカウンタループは、sorted(totals.items(), key=lambda kv: kv[1], reverse=True)[:n] の1表現で置き換えられる。スライスは n が要素数を超えても IndexError にならず、境界チェックが自動で行われる。
3
or DEFAULT_CATEGORY で None/空文字列を一括正規化(Ch2)if cat == None: cat = "unknown" は①== ではなく is None が正しいイディオム、②"unknown" というマジックストリングが散在するという2つの問題を抱える。item.get("category") or DEFAULT_CATEGORY の1行で None も空文字列も一括処理でき、定数 DEFAULT_CATEGORY に名前を付けることで意図が明確になる。
4
金額に
Decimal(str(price)) で金額精度を保証(Ch2)金額に
float を使うと 2500 * 3 * 0.1 = 749.9999... のような浮動小数点誤差が発生する。Decimal(str(price)) で float → str → Decimal と変換することで金額計算の精度を保持できる(Decimal(float) は精度が失われるため str 経由が必須)。
5
入力を
@dataclass(frozen=True, slots=True) で値オブジェクト化(Ch3)入力を
dict のままで触ると item["categorry"] のようなタイポが実行時まで検出されない。OrderItem dataclass にすると item.category と属性アクセスになり mypy で検出できる。frozen=True で不変性を保証し、slots=True でメモリ効率化とタイポによる属性追加を防ぐ。
6
subtotal プロパティに計算ロジックを局所化(Ch3/Ch6)item["price"] * item["qty"] という計算が集計関数の中に直接書かれていると、将来「割引率適用後の小計」に変えたいとき複数箇所を変更しなければならない。OrderItem.subtotal プロパティに閉じ込めると変更箇所が1箇所になり(SRP)、テストも容易になる。
実務への応用
- MOpsチームの販促集計バッチ: 「カテゴリ別クーポン利用額ランキング」を算出するシナリオはまさにこのパターン。
defaultdictによる集計ロジックは BigQuery / dbt で書くとSUM() OVER (PARTITION BY category)に対応する。Python バッチ側でも同じ設計原則(名前付き定数・値オブジェクト・型安全コレクション操作)を適用することで、バグ混入率を下げつつ可読性の高いコードが書ける。 - DataDog カスタムメトリクス:
OrderItem.subtotalプロパティをそのまま使い、カテゴリ別売上を DataDog にカスタムメトリクスとして送信できる。frozen=Truedataclass はスレッドセーフで、並行バッチ処理での共有データとして安全に使える。 - Argo Workflows のバッチステップ:
parse_order → aggregate_by_category → get_top_categoriesの3段階を Argo の逐次ステップとして分離しやすい構造になっている。各ステップの入出力が型付き値オブジェクトであるため、インターフェースが明確で独立テストが書きやすい。
今日のまとめ
defaultdict + sorted(..., reverse=True)[:n] でコレクション操作を整理し、frozen=True, slots=True dataclass で値オブジェクト化することで、型安全・簡潔・テスタブルな集計ロジックが実現できる。Decimal(str(price)) による金額精度の保証と or DEFAULT_CATEGORY による None 正規化も、ECサイト・MOps 文脈では実務必須の習慣。カテゴリの不明値を黙って捨てず "unknown" として集計することが、データ品質の可観測性につながる。