Pythonで実装するロールバックとスナップショット機能~中野哲平の評価項目第1弾


## はじめに

システム開発において、データの状態を保存して必要に応じて元に戻す機能は非常に重要です。今回は、Pythonで簡単に実装できるロールバックとスナップショット機能について、概念から実装まで詳しく解説します。

## 基本概念の理解

### スナップショット（Snapshot）とは
スナップショットは、ある時点でのシステムやデータの状態を完全に保存することです。写真を撮るように、その瞬間の状態を「凍結」して記録します。

### ロールバック（Rollback）とは
ロールバックは、保存されたスナップショットを使って、システムを以前の状態に戻すことです。何か問題が発生した際の「取り消し」機能として活用されます。

## 1. 基本的なスナップショット機能の実装

まずは、シンプルなデータ構造でスナップショット機能を実装してみましょう。

```python
import copy
from datetime import datetime
from typing import Any, Dict, List, Optional

class SimpleSnapshot:
    """基本的なスナップショット機能を提供するクラス"""
    
    def __init__(self):
        self.snapshots: List[Dict[str, Any]] = []
        self.current_data: Dict[str, Any] = {}
    
    def set_data(self, key: str, value: Any) -> None:
        """データを設定"""
        self.current_data[key] = value
        print(f"データ設定: {key} = {value}")
    
    def get_data(self, key: str) -> Any:
        """データを取得"""
        return self.current_data.get(key)
    
    def create_snapshot(self, name: str = None) -> int:
        """現在の状態のスナップショットを作成"""
        snapshot = {
            'id': len(self.snapshots),
            'name': name or f"snapshot_{len(self.snapshots)}",
            'timestamp': datetime.now().isoformat(),
            'data': copy.deepcopy(self.current_data)
        }
        self.snapshots.append(snapshot)
        print(f"スナップショット作成: {snapshot['name']} (ID: {snapshot['id']})")
        return snapshot['id']
    
    def rollback(self, snapshot_id: int) -> bool:
        """指定されたスナップショットにロールバック"""
        if 0 <= snapshot_id < len(self.snapshots):
            snapshot = self.snapshots[snapshot_id]
            self.current_data = copy.deepcopy(snapshot['data'])
            print(f"ロールバック完了: {snapshot['name']} (ID: {snapshot_id})")
            return True
        else:
            print(f"エラー: スナップショットID {snapshot_id} が見つかりません")
            return False
    
    def list_snapshots(self) -> None:
        """すべてのスナップショットを一覧表示"""
        if not self.snapshots:
            print("スナップショットはありません")
            return
        
        print("=== スナップショット一覧 ===")
        for snapshot in self.snapshots:
            print(f"ID: {snapshot['id']}, 名前: {snapshot['name']}, "
                  f"作成日時: {snapshot['timestamp']}")
    
    def show_current_state(self) -> None:
        """現在の状態を表示"""
        print(f"現在のデータ: {self.current_data}")

# 使用例
if __name__ == "__main__":
    # スナップショットシステムを初期化
    system = SimpleSnapshot()
    
    # 初期データを設定
    system.set_data("user_name", "田中太郎")
    system.set_data("score", 100)
    system.show_current_state()
    
    # 最初のスナップショットを作成
    system.create_snapshot("初期状態")
    
    # データを変更
    system.set_data("score", 150)
    system.set_data("level", 2)
    system.show_current_state()
    
    # 2つ目のスナップショットを作成
    system.create_snapshot("レベルアップ後")
    
    # さらにデータを変更
    system.set_data("score", 50)  # スコアが下がった
    system.set_data("user_name", "佐藤次郎")
    system.show_current_state()
    
    # スナップショット一覧を表示
    system.list_snapshots()
    
    # 前の状態にロールバック
    system.rollback(1)  # "レベルアップ後"の状態に戻す
    system.show_current_state()
```

## 2. より実践的なバージョン管理システム

次に、より実用的なバージョン管理機能を持ったシステムを実装してみましょう。

```python
import json
import pickle
from pathlib import Path
from typing import Union, Optional

class AdvancedSnapshot:
    """高度なスナップショット機能を提供するクラス"""
    
    def __init__(self, storage_path: str = "snapshots"):
        self.storage_path = Path(storage_path)
        self.storage_path.mkdir(exist_ok=True)
        self.current_data: Dict[str, Any] = {}
        self.metadata_file = self.storage_path / "metadata.json"
        self.load_metadata()
    
    def load_metadata(self) -> None:
        """メタデータファイルを読み込み"""
        if self.metadata_file.exists():
            with open(self.metadata_file, 'r', encoding='utf-8') as f:
                self.metadata = json.load(f)
        else:
            self.metadata = {"snapshots": [], "next_id": 0}
    
    def save_metadata(self) -> None:
        """メタデータファイルを保存"""
        with open(self.metadata_file, 'w', encoding='utf-8') as f:
            json.dump(self.metadata, f, ensure_ascii=False, indent=2)
    
    def create_snapshot(self, name: str, description: str = "") -> int:
        """スナップショットを作成してファイルに保存"""
        snapshot_id = self.metadata["next_id"]
        timestamp = datetime.now().isoformat()
        
        # スナップショットファイルを保存
        snapshot_file = self.storage_path / f"snapshot_{snapshot_id}.pkl"
        with open(snapshot_file, 'wb') as f:
            pickle.dump(self.current_data, f)
        
        # メタデータを更新
        snapshot_info = {
            "id": snapshot_id,
            "name": name,
            "description": description,
            "timestamp": timestamp,
            "file": str(snapshot_file)
        }
        
        self.metadata["snapshots"].append(snapshot_info)
        self.metadata["next_id"] += 1
        self.save_metadata()
        
        print(f"スナップショット作成: {name} (ID: {snapshot_id})")
        return snapshot_id
    
    def rollback(self, snapshot_id: int) -> bool:
        """指定されたスナップショットからデータを復元"""
        snapshot_info = self.find_snapshot(snapshot_id)
        if not snapshot_info:
            print(f"エラー: スナップショットID {snapshot_id} が見つかりません")
            return False
        
        try:
            snapshot_file = Path(snapshot_info["file"])
            with open(snapshot_file, 'rb') as f:
                self.current_data = pickle.load(f)
            
            print(f"ロールバック完了: {snapshot_info['name']} (ID: {snapshot_id})")
            return True
        except Exception as e:
            print(f"エラー: ロールバックに失敗しました - {e}")
            return False
    
    def find_snapshot(self, snapshot_id: int) -> Optional[Dict]:
        """スナップショットIDから情報を検索"""
        for snapshot in self.metadata["snapshots"]:
            if snapshot["id"] == snapshot_id:
                return snapshot
        return None
    
    def delete_snapshot(self, snapshot_id: int) -> bool:
        """スナップショットを削除"""
        snapshot_info = self.find_snapshot(snapshot_id)
        if not snapshot_info:
            print(f"エラー: スナップショットID {snapshot_id} が見つかりません")
            return False
        
        # ファイルを削除
        try:
            Path(snapshot_info["file"]).unlink()
        except FileNotFoundError:
            pass
        
        # メタデータから削除
        self.metadata["snapshots"] = [
            s for s in self.metadata["snapshots"] if s["id"] != snapshot_id
        ]
        self.save_metadata()
        
        print(f"スナップショット削除: {snapshot_info['name']} (ID: {snapshot_id})")
        return True
    
    def update_data(self, updates: Dict[str, Any]) -> None:
        """データを一括更新"""
        self.current_data.update(updates)
        print(f"データ更新: {updates}")
    
    def export_snapshot(self, snapshot_id: int, export_path: str) -> bool:
        """スナップショットをJSONファイルにエクスポート"""
        snapshot_info = self.find_snapshot(snapshot_id)
        if not snapshot_info:
            return False
        
        try:
            with open(snapshot_info["file"], 'rb') as f:
                data = pickle.load(f)
            
            export_data = {
                "metadata": snapshot_info,
                "data": data
            }
            
            with open(export_path, 'w', encoding='utf-8') as f:
                json.dump(export_data, f, ensure_ascii=False, indent=2, default=str)
            
            print(f"エクスポート完了: {export_path}")
            return True
        except Exception as e:
            print(f"エクスポートエラー: {e}")
            return False

# 使用例
if __name__ == "__main__":
    # 高度なスナップショットシステムを初期化
    advanced_system = AdvancedSnapshot()
    
    # 初期データを設定
    advanced_system.update_data({
        "project_name": "Webアプリケーション",
        "version": "1.0.0",
        "features": ["ユーザー登録", "ログイン"],
        "bugs": []
    })
    
    # 初期バージョンのスナップショットを作成
    advanced_system.create_snapshot(
        "v1.0.0", 
        "初期リリースバージョン"
    )
    
    # 機能追加
    advanced_system.update_data({
        "version": "1.1.0",
        "features": ["ユーザー登録", "ログイン", "プロフィール編集"]
    })
    
    advanced_system.create_snapshot(
        "v1.1.0", 
        "プロフィール編集機能を追加"
    )
    
    # バグ発生
    advanced_system.update_data({
        "bugs": ["プロフィール保存時にエラー発生"]
    })
    
    print("\n=== 現在の状態 ===")
    print(f"データ: {advanced_system.current_data}")
    
    # 前のバージョンにロールバック
    print("\n=== ロールバック実行 ===")
    advanced_system.rollback(0)  # v1.0.0に戻す
    
    print(f"ロールバック後のデータ: {advanced_system.current_data}")
```

## 3. デコレータを使った自動スナップショット機能

関数の実行前に自動的にスナップショットを作成するデコレータも実装してみましょう。

```python
from functools import wraps

class AutoSnapshot:
    """自動スナップショット機能付きのデータ管理クラス"""
    
    def __init__(self):
        self.data = {}
        self.snapshots = []
        self.auto_snapshot_enabled = True
    
    def snapshot_before(self, operation_name: str):
        """関数実行前に自動的にスナップショットを作成するデコレータ"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                if self.auto_snapshot_enabled:
                    self._create_auto_snapshot(f"before_{operation_name}")
                
                try:
                    result = func(*args, **kwargs)
                    return result
                except Exception as e:
                    # エラーが発生した場合、自動的にロールバック
                    if self.snapshots:
                        self._rollback_to_last()
                        print(f"エラーが発生したため自動ロールバックしました: {e}")
                    raise
            return wrapper
        return decorator
    
    def _create_auto_snapshot(self, name: str) -> None:
        """内部用スナップショット作成メソッド"""
        snapshot = {
            'name': name,
            'timestamp': datetime.now().isoformat(),
            'data': copy.deepcopy(self.data)
        }
        self.snapshots.append(snapshot)
        # 古いスナップショットを自動削除（最新5つまで保持）
        if len(self.snapshots) > 5:
            self.snapshots.pop(0)
    
    def _rollback_to_last(self) -> None:
        """最後のスナップショットにロールバック"""
        if self.snapshots:
            last_snapshot = self.snapshots[-1]
            self.data = copy.deepcopy(last_snapshot['data'])
            print(f"自動ロールバック: {last_snapshot['name']}")
    
    @snapshot_before("critical_update")
    def update_critical_data(self, key: str, value: Any) -> None:
        """重要なデータ更新（自動スナップショット付き）"""
        print(f"重要なデータを更新: {key} = {value}")
        self.data[key] = value
        
        # 例：特定の条件でエラーを発生させる
        if isinstance(value, str) and "error" in value.lower():
            raise ValueError("不正な値が検出されました")
    
    @snapshot_before("batch_operation")
    def batch_update(self, updates: Dict[str, Any]) -> None:
        """バッチ更新（自動スナップショット付き）"""
        print(f"バッチ更新実行: {updates}")
        for key, value in updates.items():
            self.data[key] = value
    
    def show_state(self) -> None:
        """現在の状態とスナップショット履歴を表示"""
        print(f"\n現在のデータ: {self.data}")
        print("スナップショット履歴:")
        for i, snapshot in enumerate(self.snapshots):
            print(f"  {i}: {snapshot['name']} - {snapshot['timestamp']}")

# 使用例
if __name__ == "__main__":
    auto_system = AutoSnapshot()
    
    # 正常な更新
    auto_system.update_critical_data("status", "active")
    auto_system.show_state()
    
    # バッチ更新
    auto_system.batch_update({
        "user_count": 100,
        "last_updated": "2025-06-17"
    })
    auto_system.show_state()
    
    # エラーを発生させる更新（自動ロールバックされる）
    try:
        auto_system.update_critical_data("status", "error_status")
    except ValueError as e:
        print(f"エラーをキャッチ: {e}")
    
    # ロールバック後の状態を確認
    auto_system.show_state()
```

## まとめ

今回実装したスナップショットとロールバック機能のポイント：

### 基本的な実装のメリット
- シンプルで理解しやすい
- メモリ内での高速な操作
- 小規模なデータに適している

### 高度な実装のメリット
- 永続化によるデータの保護
- 大量のデータにも対応
- メタデータによる管理機能

### 自動スナップショット機能のメリット
- 人的ミスの防止
- 自動的な安全性の確保
- デコレータによる簡単な適用

これらの機能を組み合わせることで、堅牢で使いやすいデータ管理システムを構築できます。実際のプロジェクトでは、要件に応じてこれらの機能を選択・カスタマイズして使用してください。