コーディング/アルゴリズム — StrEnum × 非破壊的更新 × リソース管理

概要

問題

以下の「悪いコード」は ECサイトの注文データを処理する Python モジュール。問題点を列挙し、第3章（クラスで表現する）・第5章（低凝集） の原則を適用してリファクタリングしてください。

制約・前提条件

• Python 3.12 以上
• items の各要素は {"price": float, "qty": int} の dict
• status は "pending" / "approved" / "review" の3値のみ
• save_orders はファイル書き込み失敗も考慮すること
• Pydantic v2 は使わず、標準ライブラリ（dataclasses, enum, pathlib）のみ使用
• process_orders は元のリストを破壊的変更しないこと

悪いコード (Before)

📋 コピー# order_processor.py
from dataclasses import dataclass
import json

@dataclass
class Order:
    id: int
    user_id: int
    items: list
    status: str

def process_orders(orders):
    result = []
    for o in orders:
        if o.status == "pending":
            total = 0
            for item in o.items:
                total = total + item["price"] * item["qty"]
            if total > 10000:
                o.status = "approved"
            else:
                o.status = "review"
            result.append(o)
    return result

def save_orders(orders, path):
    data = []
    for o in orders:
        data.append({
            "id": o.id,
            "user_id": o.user_id,
            "items": o.items,
            "status": o.status
        })
    f = open(path, "w")
    json.dump(data, f)
    f.close()
    print("saved")

ヒント（段階的開示）

def process_orders(orders):
    for o in orders:
        if o.status == "pending":
            # 呼び出し元のオブジェクトを直接変更
            o.status = "approved"
            result.append(o)

def _classify_order(order: Order) -> Order:
    total = _calculate_total(order)
    new_status = (
        OrderStatus.APPROVED
        if total >= APPROVAL_THRESHOLD
        else OrderStatus.REVIEW
    )
    # 新しいインスタンスを返す
    return dataclasses.replace(order, status=new_status)

問題点分析

設計構造図 — 改善後のモジュール構成

模範解答

📋 コピー# order_processor.py
"""注文データの処理と永続化モジュール。

このモジュールは pending 状態の注文を合計金額で審査し、
JSON ファイルへ書き出す機能を提供します。

Typical usage example::

    orders = [Order(id=1, user_id=42, items=[OrderItem(price=5000.0, qty=3)],
                    status=OrderStatus.PENDING)]
    processed = process_orders(orders)
    save_orders(processed, Path("orders.json"))
"""

from __future__ import annotations

import dataclasses
import json
import logging
from dataclasses import dataclass
from enum import StrEnum
from pathlib import Path
from typing import Final

logger = logging.getLogger(__name__)

# --- 定数 ---
APPROVAL_THRESHOLD: Final[float] = 10_000.0  # この金額以上で自動承認


class OrderStatus(StrEnum):
    """注文ステータスの列挙型。

    文字列との互換性を保ちつつ、タイポによる不正値を防ぐ。
    """

    PENDING = "pending"
    APPROVED = "approved"
    REVIEW = "review"


@dataclass
class OrderItem:
    """注文明細の1行。"""

    price: float
    qty: int

    def subtotal(self) -> float:
        """小計を返す。"""
        return self.price * self.qty


@dataclass
class Order:
    """注文エンティティ。

    Attributes:
        id: 注文ID。
        user_id: 購入ユーザーのID。
        items: 注文明細のリスト。
        status: 現在のステータス。
    """

    id: int
    user_id: int
    items: list[OrderItem]
    status: OrderStatus


def _calculate_total(order: Order) -> float:
    """注文合計金額を計算する。

    Args:
        order: 対象の注文。

    Returns:
        全明細の合計金額。
    """
    return sum(item.subtotal() for item in order.items)


def _classify_order(order: Order) -> Order:
    """pending 注文を合計金額で審査し、新しい Order を返す（非破壊的）。

    Args:
        order: pending 状態の注文。

    Returns:
        status が approved または review に更新された新しい Order。
    """
    total = _calculate_total(order)
    new_status = (
        OrderStatus.APPROVED if total >= APPROVAL_THRESHOLD else OrderStatus.REVIEW
    )
    # dataclasses.replace で元オブジェクトを変更せず新しいインスタンスを生成
    return dataclasses.replace(order, status=new_status)


def process_orders(orders: list[Order]) -> list[Order]:
    """pending 状態の注文を審査して返す。

    呼び出し元のリストおよびオブジェクトを破壊的変更しない。

    Args:
        orders: 処理対象の注文リスト（mixed status 可）。

    Returns:
        pending だった注文を審査済みステータスで返したリスト。
    """
    return [
        _classify_order(order)
        for order in orders
        if order.status == OrderStatus.PENDING
    ]


def save_orders(orders: list[Order], path: Path) -> None:
    """注文リストを JSON ファイルへ書き出す。

    Args:
        orders: 書き出す注文リスト。
        path: 保存先ファイルパス。

    Raises:
        OSError: ファイルの書き込みに失敗した場合。
    """
    records = [dataclasses.asdict(order) for order in orders]
    try:
        # コンテキストマネージャで確実にファイルをクローズ
        with path.open("w", encoding="utf-8") as f:
            json.dump(records, f, ensure_ascii=False, indent=2)
        logger.info("Saved %d orders to %s", len(orders), path)
    except OSError:
        logger.exception("Failed to save orders to %s", path)
        raise  # 呼び出し元に例外を伝搬させ、サイレント失敗を防ぐ

ポイント解説

Before vs After 比較

実務への応用

• Argo Workflows との親和性: 副作用フリーな関数はべき等性を保証しやすく、ワークフローの再実行（retry）に対して安全
• GCS URI との互換: save_orders の path を Path 型にしておくと PurePosixPath でラップしたGCSユーティリティへ差し替えやすい
• BigQuery との親和性: OrderStatus を StrEnum にしておくと BigQuery の STRING カラムとのマッピングが自然に行える

今日のまとめ

自己評価

自分の回答

気づき・メモ