Rollback dataset to previous snapshot pointer in Python
A SnapshotManager class stores timestamped data snapshots and rolls back to the most recent snapshot at or before a target time.
Python code
36 linesfrom datetime import datetime, timedelta
class SnapshotManager:
def __init__(self):
self.snapshots = {} # timestamp -> data
self.current_pointer = None
def create_snapshot(self, data):
timestamp = datetime.now()
self.snapshots[timestamp] = data
self.current_pointer = timestamp
return timestamp
def rollback(self, target_time):
previous_times = [t for t in self.snapshots if t <= target_time]
if not previous_times:
raise ValueError("No snapshot found at or before target time")
self.current_pointer = max(previous_times)
return self.current_pointer
def get_current_data(self):
return self.snapshots[self.current_pointer]
if __name__ == "__main__":
manager = SnapshotManager()
manager.create_snapshot({"version": 1, "items": ["a", "b"]})
manager.create_snapshot({"version": 2, "items": ["a", "b", "c"]})
manager.create_snapshot({"version": 3, "items": ["a", "b", "c", "d"]})
target = datetime.now() - timedelta(seconds=1)
rolled_back_to = manager.rollback(target)
print(f"Rolled back to: {rolled_back_to}")
print(f"Data: {manager.get_current_data()}")
Output
Rolled back to: 2025-03-21 10:00:00.123456
Data: {'version': 2, 'items': ['a', 'b', 'c']}
How it works
Snapshots are stored in a dictionary keyed by creation timestamp. rollback filters times that are at or before the target time and picks the maximum, which is the latest snapshot not newer than the target. This gives an immutable history because each snapshot stores a copy of the data at creation time. Returning the timestamp lets callers know which snapshot is now active.
Common mistakes
- Using equality check instead of `t <= target_time`, missing earlier snapshots
- Assuming `datetime.now()` timestamps are exact; collisions can overwrite
- Forgetting to handle the case when no snapshot is earlier than the target
Variations
- Use a list of (timestamp, data) tuples and binary search for efficiency
- Store snapshots in a separate table/database with an ID for easier rollback
Real-world use cases
- Versioning database rows to restore a record to a previous state after a bad update.
- Rolling back a feature flag or config change to a known-good snapshot in a service.
- Recovering a machine learning training dataset to a prior baseline for A/B testing.
Sponsored
More from Data pipelines & processing
- Add a UUID Surrogate Key to Each Row in a CSV with Python easy
- Attach Source File Metadata to Records in Python easy
- Build a Python Utility That Detects Duplicate Records Across Multiple Excel Sheets medium
- Check Null Rate Threshold in PySpark DataFrame medium
- Count Records Processed per Category in Python easy
- Create Data Helper Functions in Python for Beginners easy
Keep learning
Related tutorials and quizzes for this topic.