備忘録: Iterator パターンでコレクション操作を一般化：CSVデータの変換処理

Posted at 2025-03-08

はじめに

Iterator パターンは、コレクションの要素に順番にアクセスする方法を提供するデザインパターンです。本記事では、CSVデータの変換処理という実用的なシナリオを通じて、Iterator パターンの有用性を説明します。

なぜ Iterator パターンを使うのか

抽象化: 複雑なデータ変換ロジックを隠蔽し、シンプルなインターフェースを提供します。
柔軟性: 異なる種類のデータ変換を容易に追加・削除できます。
単一責任の原則: データの保持、変換、イテレーションの責務を分離します。

使い所

大量のCSVデータを処理する必要がある場合
データクレンジングや前処理のパイプラインを構築する場合
異なるデータフィールドに対して異なる変換を適用する場合

メリット

変換処理の追加・削除が容易
メモリ効率の良い遅延評価が可能
コードの再利用性と保守性が向上

デメリット

単純なデータ処理の場合、実装が複雑になる可能性がある
イテレータの状態管理に注意が必要

実装例

以下に、CSVデータを処理し、各フィールドに対して異なる変換を適用するカスタムイテレータの実装例を示します。

from abc import ABC, abstractmethod
from typing import Callable, Dict
import csv
from datetime import datetime

# Iterator インターフェース
class Iterator(ABC):
    @abstractmethod
    def __iter__(self):
        pass

    @abstractmethod
    def __next__(self) -> Dict[str, str]:
        pass

# Aggregate インターフェース
class Aggregate(ABC):
    @abstractmethod
    def __iter__(self) -> Iterator:
        pass

# 具体的な Aggregate クラス
class CSVTransformer(Aggregate):
    def __init__(self, file_path: str):
        self._file_path = file_path
        self._transformations: Dict[str, Callable[[str], str]] = {}

    def add_transformation(self, field: str, transformation: Callable[[str], str]):
        self._transformations[field] = transformation

    def __iter__(self) -> Iterator:
        return CSVTransformerIterator(self)

# 具体的な Iterator クラス
class CSVTransformerIterator(Iterator):
    def __init__(self, csv_transformer: CSVTransformer):
        self._csv_transformer = csv_transformer
        self._file = open(self._csv_transformer._file_path, 'r', newline='', encoding='utf-8')
        self._csv_reader = csv.DictReader(self._file)
        self._iter = iter(self._csv_reader)

    def __iter__(self):
        return self

    def __next__(self) -> Dict[str, str]:
        row = next(self._iter)  # StopIteration が発生する可能性があります
        transformed_row = {}
        for field, value in row.items():
            if field in self._csv_transformer._transformations:
                try:
                    transformed_row[field] = self._csv_transformer._transformations[field](value)
                except Exception as e:
                    transformed_row[field] = value  # エラー時は元の値を保持
                    print(f"変換エラー（フィールド: {field}, 値: {value}）: {e}")
            else:
                transformed_row[field] = value
        return transformed_row

    def __del__(self):
        if self._file:
            self._file.close()

# 変換処理の定義
def capitalize_name(name: str) -> str:
    return name.title()

def format_date(date_str: str) -> str:
    date = datetime.strptime(date_str, "%Y-%m-%d")
    return date.strftime("%B %d, %Y")

def convert_salary(salary_str: str) -> str:
    salary = float(salary_str)
    return f"${salary:,.2f}"

# テスト用のCSVファイルを作成
def create_test_csv(file_path: str):
    data = [
        {"name": "john doe", "hire_date": "2022-03-15", "salary": "75000"},
        {"name": "jane smith", "hire_date": "2021-11-01", "salary": "82000"},
        {"name": "bob johnson", "hire_date": "2023-01-10", "salary": "68000"},
    ]
    fieldnames = ["name", "hire_date", "salary"]
    with open(file_path, 'w', newline='', encoding='utf-8') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(data)

# 使用例
if __name__ == "__main__":
    csv_file = "employees.csv"
    create_test_csv(csv_file)  # テスト用CSVファイルの作成

    transformer = CSVTransformer(csv_file)
    
    # 変換処理の追加
    transformer.add_transformation("name", capitalize_name)
    transformer.add_transformation("hire_date", format_date)
    transformer.add_transformation("salary", convert_salary)

    print("変換後のデータ:")
    for row in transformer:
        print(row)

ポイント

テスト用CSVファイルの作成: create_test_csv 関数を追加し、実行時にテスト用のCSVファイルを自動生成します。これにより、コードをそのまま実行して動作を確認できます。
Pythonのイテレータプロトコルに準拠: __iter__() と __next__() メソッドを実装し、Pythonの組み込みイテレータプロトコルに対応しました。これにより、for ループで直接イテレータを使用できます。
エラーハンドリングの強化: 変換処理中に例外が発生した場合でも、処理を継続し、エラーメッセージを表示します。
ファイルのエンコーディングと改行コードの指定: ファイル操作時に encoding='utf-8' と newline='' を指定し、異なるプラットフォーム間での互換性を高めました。

実行結果

このコードを実行すると、以下のような出力が得られます：

変換後のデータ:
{'name': 'John Doe', 'hire_date': 'March 15, 2022', 'salary': '$75,000.00'}
{'name': 'Jane Smith', 'hire_date': 'November 01, 2021', 'salary': '$82,000.00'}
{'name': 'Bob Johnson', 'hire_date': 'January 10, 2023', 'salary': '$68,000.00'}

各行に対して、以下の変換が適用されています：

name: 各単語の先頭文字を大文字に変換
hire_date: 日付形式を "YYYY-MM-DD" から "Month DD, YYYY" に変更
salary: 数値を通貨形式（例: $75,000.00）に変換

まとめ

この例では、Iterator パターンを使用してCSVデータの変換処理を実装しました。この方法には以下の利点があります：

新しい変換処理の追加が容易: Open-Closed 原則に準拠しており、既存のコードを変更せずに新しい機能を追加できます。
大量のデータに対しても効率的に処理可能: 遅延評価により、メモリ使用量を最小限に抑えられます。
各フィールドに対して異なる変換を適用可能: フィールドごとに個別の変換ロジックを設定できます。

Iterator パターンは、このようなデータ変換処理のシナリオで特に有用です。大量のデータを処理する際のメモリ効率、変換ロジックの柔軟な追加・変更、コードの再利用性向上などの利点があります。

ただし、データ量が少ない場合や、変換処理が単純な場合は、このパターンが過剰になる可能性があります。使用する際は、プロジェクトの要件と規模を考慮して判断してください。

追加の注意点

リソース管理: __del__() メソッドでのファイルクローズは、タイミングが不定になる可能性があります。より確実にファイルをクローズするために、contextlib モジュールの contextmanager デコレータや with 文を使用することも検討してください。
エラーハンドリング: 実運用環境では、データの不整合や予期しない入力に対するエラーハンドリングが重要です。必要に応じて例外処理を強化してください。
コードの最適化: Pythonの標準ライブラリやジェネレータを活用することで、コードをさらに簡潔かつ効率的にすることができます。

Iterator パターンを活用することで、柔軟で保守性の高いデータ変換処理を実装できます。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up