How to Take Periodic Snapshots of Aggregate State in Python

Build a Python class that accumulates values and periodically captures immutable snapshots of total, count, and average for later analysis.

Easy Python 3.9+ Aug 9, 2026 System design patterns 12 views 0 copies

aggregation snapshots state-management system-design oop

Python code

37 lines

Python 3.9+

import time
import random
from collections import defaultdict


class SnapshotAggregator:
    def __init__(self):
        self.total = 0
        self.count = 0
        self.history = []

    def add(self, value):
        self.total += value
        self.count += 1

    def snapshot(self):
        avg = self.total / self.count if self.count else 0
        state = {
            "total": self.total,
            "count": self.count,
            "average": round(avg, 2),
        }
        self.history.append(state.copy())
        return state


if __name__ == "__main__":
    agg = SnapshotAggregator()
    random.seed(42)

    for tick in range(5):
        agg.add(random.randint(1, 10))
        print(f"tick={tick} -> {agg.snapshot()}")

    print("history:")
    for state in agg.history:
        print(state)

Output

stdout

tick=0 -> {'total': 2, 'count': 1, 'average': 2.0}
tick=1 -> {'total': 11, 'count': 2, 'average': 5.5}
tick=2 -> {'total': 16, 'count': 3, 'average': 5.33}
tick=3 -> {'total': 22, 'count': 4, 'average': 5.5}
tick=4 -> {'total': 25, 'count': 5, 'average': 5.0}
history:
{'total': 2, 'count': 1, 'average': 2.0}
{'total': 11, 'count': 2, 'average': 5.5}
{'total': 16, 'count': 3, 'average': 5.33}
{'total': 22, 'count': 4, 'average': 5.5}
{'total': 25, 'count': 5, 'average': 5.0}

How it works

The SnapshotAggregator maintains incremental totals and counts, so each add call is O(1). The snapshot method computes the average on demand and stores a copy of the state dict, preventing later mutations from corrupting history. The code uses random.seed(42) to produce deterministic output, which is useful for tests and demos. The history list keeps references to immutable snapshots, making it safe to iterate over later without side effects. This pattern decouples the live aggregation state from periodic reporting, a common design in metrics and observability systems.

Common mistakes

Storing the same dict object in history without copying, so later snapshots overwrite earlier entries
Dividing by zero when no values have been added yet, leading to ZeroDivisionError
Using mutable defaults in __init__ (e.g., def __init__(self, history=[])) which share state across instances

Variations

Use `collections.deque` with maxlen to keep only the latest N snapshots
Implement `__getitem__` or a generator to iterate over history lazily

Real-world use cases

Aggregating API request counts and latencies each minute for a monitoring dashboard.
Collecting totals in a data pipeline and emitting periodic batch snapshots for downstream reporting.
Tracking rolling averages of sensor readings in an IoT gateway and logging snapshots to cloud storage.

How to Take Periodic Snapshots of Aggregate State in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from System design patterns

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from System design patterns

Keep learning

Tutorials

Quizzes