The Memento pattern captures an object's internal state into a separate snapshot object, so the object can be restored to that state later without exposing its internals to the rest of the program. The three roles are the originator (the object with state), the memento (the opaque snapshot), and the caretaker (whatever holds onto snapshots and decides when to restore). In Python the snapshot is usually a small immutable record, and the originator knows how to produce one and load one back.

A checkpointable pipeline stage

The pattern fits any long-running process that needs to roll back: an editor's undo stack, a state machine, or a data pipeline that checkpoints progress so a crash doesn't restart from zero. Here a stateful aggregator can save a memento before a risky batch and restore it if the batch fails. The memento is a frozen dataclass, so the caretaker can hold it but can't mutate the originator through it.

from dataclasses import dataclass, field
from copy import deepcopy

@dataclass(frozen=True)
class Memento:
    processed: int
    totals: dict[str, float]

@dataclass
class Aggregator:
    processed: int = 0
    totals: dict[str, float] = field(default_factory=dict)

    def add(self, key: str, value: float) -> None:
        self.totals[key] = self.totals.get(key, 0.0) + value
        self.processed += 1

    def save(self) -> Memento:
        return Memento(self.processed, deepcopy(self.totals))

    def restore(self, m: Memento) -> None:
        self.processed = m.processed
        self.totals = deepcopy(m.totals)

agg = Aggregator()
agg.add("rev", 100.0)
checkpoint = agg.save()          # snapshot before a risky batch
try:
    agg.add("rev", 50.0)
    raise RuntimeError("batch failed")
except RuntimeError:
    agg.restore(checkpoint)      # roll back

print(agg.processed, agg.totals)  # 1 {'rev': 100.0}

The caretaker and an undo history

The caretaker owns the snapshots and never inspects their contents - it just stacks and pops them. A bounded history gives undo with a memory ceiling, which matters when each snapshot is a sizable pipeline state. Keep the deep copy on the originator's side so the caretaker can't accidentally share mutable structure between snapshots.

class History:
    def __init__(self, origin: Aggregator, limit: int = 50):
        self.origin = origin
        self.limit = limit
        self._stack: list[Memento] = []

    def commit(self) -> None:
        self._stack.append(self.origin.save())
        if len(self._stack) > self.limit:
            self._stack.pop(0)

    def undo(self) -> bool:
        if not self._stack:
            return False
        self.origin.restore(self._stack.pop())
        return True

h = History(agg)
h.commit()
agg.add("rev", 999.0)
h.undo()
print(agg.totals)  # {'rev': 100.0}

Serialization is the durable variant

When the snapshot has to survive a process restart - the real checkpoint case - the memento becomes a serialized blob on disk or in a store rather than an in-memory object. The same roles hold: the originator decides what goes into the snapshot and how to rebuild from it, and the caretaker (now a file or a key in object storage) just keeps the bytes. Convert the frozen record to JSON, Parquet, or pickle on save and parse it back on restore; the rest of the program still never sees the originator's fields directly.

import json
from dataclasses import asdict

def write_checkpoint(path: str, m: Memento) -> None:
    with open(path, "w", encoding="utf-8") as f:
        json.dump(asdict(m), f)

def read_checkpoint(path: str) -> Memento:
    with open(path, encoding="utf-8") as f:
        return Memento(**json.load(f))

Frequently asked questions

What's the difference between the memento and just copying the object? The memento is a deliberate, often minimal snapshot of the state worth saving, and it's opaque to the caretaker. A blind copy exposes everything and couples the holder to the object's internals.

Why deep-copy the state? So restoring an old snapshot doesn't share mutable containers (dicts, lists) with the live object - otherwise a later mutation would corrupt the saved state. Skip the copy only for immutable fields.

How does this relate to pipeline checkpointing? A checkpoint is a durable memento: the pipeline stage serializes enough state (offsets, partial aggregates, cursor) to resume after a failure without reprocessing from the start.

Won't keeping many mementos use a lot of memory? It can, which is why the caretaker usually bounds the history or persists older snapshots to disk and keeps only recent ones in memory.