Pythonで実装するVisitorパターンによるデータ変換処理

## はじめに

JSONデータを異なる形式（CSV、YAML、XML）に変換する問題をテーマとします。この記事では、Visitorパターンを使用して柔軟で拡張性の高いデータ変換処理を実装する方法を解説します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/d45ec218-23a9-3953-41d3-399a843134c0.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/26e47a2d-fcd1-9154-1e6e-aef1f79a9165.png)


## なぜVisitorパターンなのか？

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/31ba1974-5ff7-c6e6-3315-ea9ed8e53da0.png)


### 従来の実装の問題点

データ変換処理でよく見られる実装方法は以下のようなものです：

```python
def convert_data(data, format_type):
    if format_type == "csv":
        return to_csv(data)
    elif format_type == "yaml":
        return to_yaml(data)
    elif format_type == "xml":
        return to_xml(data)
    # 新しい形式を追加するたびにif文が増える...
```

この実装には以下の問題があります：

1. **拡張性の問題**: 新しい形式を追加するたびに既存コードを修正
2. **保守性の問題**: 各変換処理のロジックが散らばる
3. **再利用性の問題**: 共通処理の再利用が難しい

### Visitorパターンのメリット

Visitorパターンを採用することで、以下のメリットが得られます：

1. **データ構造と処理の分離**: データ構造自体に変更を加えずに新しい処理を追加可能
2. **拡張性の向上**: 新しい変換処理は新しいVisitorクラスを追加するだけ
3. **責務の明確化**: 各変換処理が専用のクラスに集約される
4. **再利用性の向上**: 共通のロジックを基底クラスで実装可能

## 実装例

### 1. データ構造の定義

まず、JSONデータを表現するクラス階層を定義します：

```python
from dataclasses import dataclass
from typing import List, Dict, Any, Union, Optional

# 型定義
JSONPrimitive = Union[str, int, float, bool, None]
JSONValue = Union['JSONNode', JSONPrimitive]

@dataclass
class JSONObject(JSONNode):
    values: Dict[str, JSONValue]
    
    def accept(self, visitor: 'JSONVisitor') -> None:
        visitor.visit_object(self)
        
@dataclass
class JSONArray(JSONNode):
    items: List[JSONValue]
    
    def accept(self, visitor: 'JSONVisitor') -> None:
        visitor.visit_array(self)
```

### 2. 変換処理の実装

各形式への変換処理を実装します。以下は主要な実装の例です：

#### CSVコンバーター

```python
class CSVConverter(JSONVisitor):
    def __init__(self):
        self.rows: List[Dict[str, str]] = []
        self.current_row: Dict[str, str] = {}
        self.headers: Set[str] = set()
    
    def _flatten_object(self, obj: JSONObject, prefix: str = '') -> None:
        for key, value in obj.values.items():
            full_key = f"{prefix}.{key}" if prefix else key
            if isinstance(value, JSONObject):
                self._flatten_object(value, full_key)
            elif isinstance(value, JSONArray):
                self._process_array(value, full_key)
            else:
                self.headers.add(full_key)
                self.current_row[full_key] = str(value)
```

#### YAMLコンバーター

```python
class YAMLConverter(JSONVisitor):
    def __init__(self):
        self.result: Any = None
        
    def _convert_value(self, value: JSONValue) -> Any:
        if isinstance(value, (str, int, float, bool)) or value is None:
            return value
        elif isinstance(value, JSONNode):
            original_result = self.result
            self.result = None
            value.accept(self)
            converted = self.result
            self.result = original_result
            return converted
        return None
```

#### XMLコンバーター

```python
class XMLConverter(JSONVisitor):
    def __init__(self):
        self.result = ""
        self.indent_level = 0
        self.indent_str = "  "
    
    def visit_object(self, obj: JSONObject) -> None:
        self.result += f"{self._indent()}<object>\n"
        self.indent_level += 1
        for key, value in obj.values.items():
            safe_key = xml_escape(key)
            self.result += f"{self._indent()}<{safe_key}>\n"
            # ... 残りの実装
```

## 変換結果の例

以下のようなサンプルデータを使用して、各形式への変換を試してみます：

```python
sample_data = {
    "users": [
        {
            "id": 1,
            "name": "山田太郎",
            "email": "yamada@example.com",
            "age": 30,
            "active": True,
            "skills": ["Python", "JavaScript", "SQL"]
        },
        {
            "id": 2,
            "name": "鈴木花子",
            "email": "suzuki@example.com",
            "age": 25,
            "active": False,
            "skills": ["Java", "C#"],
            "department": "開発部"
        }
    ]
}
```

### CSV出力結果

```csv
active,age,department,email,id,name,skills[0],skills[1],skills[2]
True,30,,yamada@example.com,1,山田太郎,Python,JavaScript,SQL
False,25,開発部,suzuki@example.com,2,鈴木花子,Java,C#,
```

CSVの特徴：
- ヘッダーがアルファベット順にソート
- 配列要素が個別のカラムに展開
- 欠損値が適切に処理

### YAML出力結果

```yaml
users:
  - id: 1
    name: 山田太郎
    email: yamada@example.com
    age: 30
    active: true
    skills:
      - Python
      - JavaScript
      - SQL
  - id: 2
    name: 鈴木花子
    email: suzuki@example.com
    age: 25
    active: false
    skills:
      - Java
      - C#
    department: 開発部
```

YAMLの特徴：
- 階層構造を保持
- データ型を適切に保持（boolean, number, string）
- 日本語を正しく処理

### XML出力結果

```xml
<?xml version="1.0" encoding="UTF-8"?>
<object>
  <users>
    <array>
      <item>
        <object>
          <id>1</id>
          <name>山田太郎</name>
          <email>yamada@example.com</email>
          <age>30</age>
          <active>True</active>
          <skills>
            <array>
              <item>Python</item>
              <item>JavaScript</item>
              <item>SQL</item>
            </array>
          </skills>
        </object>
      </item>
      <!-- ... 2番目のユーザー情報 ... -->
    </array>
  </users>
</object>
```

XMLの特徴：
- 適切なインデントで整形
- 階層構造を明確に表現
- XML宣言を含む

## 実装のポイント

1. **型安全性**
   - 型ヒントを活用
   - dataclassでの構造定義

2. **データ構造の処理**
   - 再帰的なデータ構造のサポート
   - 配列要素の適切な処理

3. **出力形式の最適化**
   - CSVのヘッダー管理
   - YAMLのデータ型保持
   - XMLの整形とエスケープ

## まとめ

Visitorパターンを使用することで：

- 新しい変換形式の追加が容易
- コードの保守性が向上
- 変換処理の再利用が可能

実装時は以下の点に注意が必要です：

- データ構造の適切な設計
- 各Visitorの責務の明確化
- エラー処理の考慮

### 実装(動作確認用)

```mermaid
classDiagram
    class JSONNode {
        <<abstract>>
        +accept(visitor: JSONVisitor)
    }
    
    class JSONObject {
        -values: Dict
        +accept(visitor: JSONVisitor)
    }
    
    class JSONArray {
        -items: List
        +accept(visitor: JSONVisitor)
    }

    class JSONVisitor {
        <<abstract>>
        +visit_object(obj: JSONObject)
        +visit_array(arr: JSONArray)
    }

    class CSVConverter {
        +visit_object(obj: JSONObject)
        +visit_array(arr: JSONArray)
        +get_result() string
    }

    class YAMLConverter {
        +visit_object(obj: JSONObject)
        +visit_array(arr: JSONArray)
        +get_result() string
    }

    class XMLConverter {
        +visit_object(obj: JSONObject)
        +visit_array(arr: JSONArray)
        +get_result() string
    }

    JSONNode <|-- JSONObject
    JSONNode <|-- JSONArray
    JSONVisitor <|-- CSVConverter
    JSONVisitor <|-- YAMLConverter
    JSONVisitor <|-- XMLConverter
    
    JSONNode ..> JSONVisitor : accepts
```

```mermaid
sequenceDiagram
    participant Client
    participant json as JSONObject
    participant array as JSONArray
    participant csv as CSVConverter
    
    note over Client,csv: JSON to CSV変換の例
    
    Client->>csv: CSVConverter生成
    Client->>json: accept(csv_converter)
    
    activate json
    json->>csv: visit_object(self)
    
    activate csv
    note over json,csv: オブジェクトのフィールドを処理
    
    csv->>json: values取得
    json-->>csv: values
    
    loop 各フィールドに対して
        alt フィールドがJSONArray
            csv->>array: accept(self)
            
            activate array
            array->>csv: visit_array(self)
            
            note over array,csv: 配列の要素を処理
            csv->>array: items取得
            array-->>csv: items
            
            deactivate array
        else フィールドが基本型
            note over csv: 値を直接CSVに変換
        end
    end
    
    deactivate csv
    deactivate json
    
    Client->>csv: get_result()
    csv-->>Client: CSV形式の文字列
```

```python
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Dict, Any, Union, Optional, TypeVar, Set
import yaml
from xml.sax.saxutils import escape as xml_escape
import csv
import io

# 型定義
JSONPrimitive = Union[str, int, float, bool, None]
JSONValue = Union['JSONNode', JSONPrimitive]
T = TypeVar('T', bound='JSONNode')

class JSONNode(ABC):
    """JSONデータの基底クラス"""
    @abstractmethod
    def accept(self, visitor: 'JSONVisitor') -> None:
        """Visitorを受け入れるメソッド"""
        pass

    def get(self, key: str, default: Any = None) -> Optional[JSONValue]:
        """キーに対応する値を取得（JSONObjectの場合）"""
        return default

@dataclass
class JSONObject(JSONNode):
    """JSONオブジェクトを表現するクラス"""
    values: Dict[str, JSONValue]
    
    def accept(self, visitor: 'JSONVisitor') -> None:
        visitor.visit_object(self)

    def get(self, key: str, default: Any = None) -> Optional[JSONValue]:
        """キーに対応する値を取得"""
        return self.values.get(key, default)

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'JSONObject':
        """通常の辞書からJSONObjectを生成"""
        values = {}
        for key, value in data.items():
            if isinstance(value, dict):
                values[key] = cls.from_dict(value)
            elif isinstance(value, list):
                values[key] = JSONArray.from_list(value)
            else:
                values[key] = value
        return cls(values)
        
@dataclass
class JSONArray(JSONNode):
    """JSON配列を表現するクラス"""
    items: List[JSONValue]
    
    def accept(self, visitor: 'JSONVisitor') -> None:
        visitor.visit_array(self)

    @classmethod
    def from_list(cls, data: List[Any]) -> 'JSONArray':
        """通常のリストからJSONArrayを生成"""
        items = []
        for value in data:
            if isinstance(value, dict):
                items.append(JSONObject.from_dict(value))
            elif isinstance(value, list):
                items.append(cls.from_list(value))
            else:
                items.append(value)
        return cls(items)

class JSONVisitor(ABC):
    """JSON変換の基底Visitorクラス"""
    @abstractmethod
    def visit_object(self, obj: JSONObject) -> None:
        """オブジェクトの処理"""
        pass
        
    @abstractmethod
    def visit_array(self, arr: JSONArray) -> None:
        """配列の処理"""
        pass

class CSVConverter(JSONVisitor):
    """CSV形式への変換を行うVisitor"""
    def __init__(self):
        self.rows: List[Dict[str, str]] = []
        self.current_row: Dict[str, str] = {}
        self.headers: Set[str] = set()
    
    def _flatten_object(self, obj: JSONObject, prefix: str = '') -> None:
        """オブジェクトをフラットな構造に変換"""
        for key, value in obj.values.items():
            full_key = f"{prefix}.{key}" if prefix else key
            if isinstance(value, JSONObject):
                self._flatten_object(value, full_key)
            elif isinstance(value, JSONArray):
                self._process_array(value, full_key)
            else:
                self.headers.add(full_key)
                self.current_row[full_key] = str(value)

    def _process_array(self, arr: JSONArray, prefix: str) -> None:
        """配列を処理し、各要素を個別の列として追加"""
        for i, item in enumerate(arr.items):
            if isinstance(item, (str, int, float, bool)) or item is None:
                self.headers.add(f"{prefix}[{i}]")
                self.current_row[f"{prefix}[{i}]"] = str(item)
            elif isinstance(item, JSONObject):
                self._flatten_object(item, f"{prefix}[{i}]")
            elif isinstance(item, JSONArray):
                self._process_array(item, f"{prefix}[{i}]")

    def visit_object(self, obj: JSONObject) -> None:
        """JSONObjectを処理"""
        self.current_row = {}
        self._flatten_object(obj)
        if self.current_row:
            self.rows.append(self.current_row.copy())

    def visit_array(self, arr: JSONArray) -> None:
        """JSONArrayを処理"""
        for item in arr.items:
            if isinstance(item, JSONObject):
                self.current_row = {}
                self._flatten_object(item)
                self.rows.append(self.current_row.copy())

    def get_result(self) -> str:
        """CSV形式の文字列を返す"""
        if not self.rows:
            return ""
        
        # ヘッダーをソート
        headers = sorted(self.headers)
        
        # CSV出力の準備
        output = io.StringIO()
        writer = csv.DictWriter(output, fieldnames=headers)
        
        # ヘッダー行とデータ行を書き込み
        writer.writeheader()
        for row in self.rows:
            writer.writerow(row)
        
        return output.getvalue()

class YAMLConverter(JSONVisitor):
    """YAML形式への変換を行うVisitor"""
    def __init__(self):
        self.result: Any = None
        
    def _convert_value(self, value: JSONValue) -> Any:
        """値をYAML互換の形式に変換"""
        if isinstance(value, (str, int, float, bool)) or value is None:
            return value
        elif isinstance(value, JSONNode):
            original_result = self.result
            self.result = None
            value.accept(self)
            converted = self.result
            self.result = original_result
            return converted
        return None

    def visit_object(self, obj: JSONObject) -> None:
        """JSONObjectを処理"""
        result = {}
        for key, value in obj.values.items():
            result[key] = self._convert_value(value)
        self.result = result
    
    def visit_array(self, arr: JSONArray) -> None:
        """JSONArrayを処理"""
        result = []
        for item in arr.items:
            converted = self._convert_value(item)
            result.append(converted)
        self.result = result
    
    def get_result(self) -> str:
        """YAML形式の文字列を返す"""
        if self.result is None:
            return ""
        return yaml.dump(self.result, allow_unicode=True, sort_keys=False)

class XMLConverter(JSONVisitor):
    """XML形式への変換を行うVisitor"""
    def __init__(self):
        self.result = ""
        self.indent_level = 0
        self.indent_str = "  "
    
    def _indent(self) -> str:
        """現在のインデントレベルに応じたインデント文字列を返す"""
        return self.indent_str * self.indent_level
    
    def visit_object(self, obj: JSONObject) -> None:
        """JSONObjectを処理"""
        self.result += f"{self._indent()}<object>\n"
        self.indent_level += 1
        for key, value in obj.values.items():
            safe_key = xml_escape(key)
            self.result += f"{self._indent()}<{safe_key}>\n"
            self.indent_level += 1
            if isinstance(value, JSONNode):
                value.accept(self)
            else:
                self.result += f"{self._indent()}{xml_escape(str(value))}\n"
            self.indent_level -= 1
            self.result += f"{self._indent()}</{safe_key}>\n"
        self.indent_level -= 1
        self.result += f"{self._indent()}</object>\n"
    
    def visit_array(self, arr: JSONArray) -> None:
        """JSONArrayを処理"""
        self.result += f"{self._indent()}<array>\n"
        self.indent_level += 1
        for item in arr.items:
            self.result += f"{self._indent()}<item>\n"
            self.indent_level += 1
            if isinstance(item, JSONNode):
                item.accept(self)
            else:
                self.result += f"{self._indent()}{xml_escape(str(item))}\n"
            self.indent_level -= 1
            self.result += f"{self._indent()}</item>\n"
        self.indent_level -= 1
        self.result += f"{self._indent()}</array>\n"
    
    def get_result(self) -> str:
        """整形されたXML文字列を返す"""
        return f'<?xml version="1.0" encoding="UTF-8"?>\n{self.result}'

def create_sample_data() -> JSONObject:
    """サンプルデータを生成する関数"""
    raw_data = {
        "users": [
            {
                "id": 1,
                "name": "山田太郎",
                "email": "yamada@example.com",
                "age": 30,
                "active": True,
                "skills": ["Python", "JavaScript", "SQL"]
            },
            {
                "id": 2,
                "name": "鈴木花子",
                "email": "suzuki@example.com",
                "age": 25,
                "active": False,
                "skills": ["Java", "C#"],
                "department": "開発部"
            }
        ]
    }
    return JSONObject.from_dict(raw_data)

def main():
    """使用例を示すメイン関数"""
    # サンプルデータの作成
    data = create_sample_data()
    
    # users配列の取得
    users = data.get("users")
    if not isinstance(users, JSONNode):
        print("Error: Invalid data structure")
        return
    
    # CSV形式に変換
    csv_converter = CSVConverter()
    users.accept(csv_converter)
    print("\nCSV Output:")
    print(csv_converter.get_result())

    # YAML形式に変換
    yaml_converter = YAMLConverter()
    data.accept(yaml_converter)
    print("\nYAML Output:")
    print(yaml_converter.get_result())
    
    # XML形式に変換
    xml_converter = XMLConverter()
    data.accept(xml_converter)
    print("\nXML Output:")
    print(xml_converter.get_result())

if __name__ == "__main__":
    main()
```

実行例: google colab
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/2b5c7d89-e655-ce2d-d4bb-ff89245b8277.png)