Event sourcing append store replay in Python

A simple in-memory event store that appends events per aggregate and replays them on demand.

Easy Python 3.9+ Aug 9, 2026 Streaming & messaging 14 views 0 copies

event-sourcing append-only replay in-memory

Python code

25 lines

Python 3.9+

import json
from collections import defaultdict


class EventStore:
    def __init__(self):
        self._events = defaultdict(list)

    def append(self, aggregate_id, event_type, data):
        event = {"type": event_type, "data": data}
        self._events[aggregate_id].append(event)

    def replay(self, aggregate_id):
        return list(self._events[aggregate_id])


if __name__ == "__main__":
    store = EventStore()

    store.append("order-1", "OrderCreated", {"amount": 100})
    store.append("order-1", "OrderPaid", {"paid": True})
    store.append("order-2", "OrderCreated", {"amount": 50})

    history = store.replay("order-1")
    print(json.dumps(history, indent=2))

Output

stdout

[
  {
    "type": "OrderCreated",
    "data": {
      "amount": 100
    }
  },
  {
    "type": "OrderPaid",
    "data": {
      "paid": true
    }
  }
]

How it works

The EventStore uses a defaultdict(list) so each aggregate ID automatically gets its own list of events. append wraps the event type and data into a dict and appends it to the aggregate's list. replay returns a shallow copy of the list, so callers can iterate without mutating the stored events. Because events are stored in append order, replay reconstructs the aggregate's state exactly as it happened.

Common mistakes

Using a plain dict and getting KeyError when the aggregate has no events yet
Forgetting to copy the list in replay and accidentally mutating the stored events
Storing only the latest state instead of an append-only log of events

Variations

Use a list of tuples `(event_type, data)` instead of dicts for a more compact representation
Persist events to a SQLite or Postgres table for durability across restarts

Real-world use cases

Reconstructing an order's state in an e-commerce backend from its event history.
Building audit logs where every state change must be replayable later.
Feeding events to projections or analytics systems from a stream source.

Event sourcing append store replay in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Keep learning

Tutorials

Quizzes