📷 Snapshots

Snapshot is a frozen state of the stream in a given point. The point of the snapshot is identified by the version.

Snapshots are a way to optimize loading a long stream of events. When a snapshot is present, it is always loaded. Then, any events newer then a snapshots are also loaded, if only they are present.

Let's say we have a stream with 100 events:

class TemperatureChanged(Event):
    reading_in_celsius: int

stream_id = StreamId(name="temperature_sensor/1")
events = [TemperatureChanged(reading_in_celsius=reading) for reading in range(100)]
event_store.append(*events, stream_id=stream_id)

loaded_events = event_store.load_stream(stream_id)
print(len(loaded_events))  # 100

We could define a snapshot event that will capture all information relevant for us:

class TemperatureSensorSnapshot(Event):
    readings_so_far: int
    last_reading_in_celsius: int

Now we can save a snapshot, using dedicated EventStore method:

last_version = loaded_events[-1].version
last_event = loaded_events[-1].event

snapshot = TemperatureSensorSnapshot(
    readings_so_far=100, last_reading_in_celsius=last_event.reading_in_celsius
)
wrapped_snapshot = WrappedEvent.wrap(snapshot, last_version)

event_store.save_snapshot(stream_id, wrapped_snapshot)

Note that version of the snapshot must equal to the version of the last event.

Now when you'll try to load the stream you'll notice the method only returns a single event and this will be our snapshot:

loaded_events = event_store.load_stream(stream_id)
print(len(loaded_events))  # 1
assert isinstance(loaded_events[-1].event, TemperatureSensorSnapshot)

Warning

Long streams are usually a sign of a poor stream design. Snapshots are an optimization that should be used only for a good reason. Use with caution!