Event Sourcing Store in Python: Append-Only Log Mock
Mock an append-only event store in Python — record events, list them, and fetch by ID using a simple list-backed class.
Python code
35 linesclass EventStore:
def __init__(self):
self._events = []
def append(self, event):
event_id = len(self._events) + 1
stored_event = {"id": event_id, "data": event}
self._events.append(stored_event)
return stored_event
def get_events(self):
return list(self._events)
def get_event(self, event_id):
for event in self._events:
if event["id"] == event_id:
return event
return None
if __name__ == "__main__":
store = EventStore()
store.append({"type": "UserCreated", "name": "Alice"})
store.append({"type": "UserUpdated", "name": "Alice", "age": 30})
event = store.append({"type": "UserDeleted", "name": "Alice"})
print("All events:")
for e in store.get_events():
print(e)
print("\nLatest event:")
print(event)
print("\nFetched event #2:")
print(store.get_event(2))
Output
All events:
{'id': 1, 'data': {'type': 'UserCreated', 'name': 'Alice'}}
{'id': 2, 'data': {'type': 'UserUpdated', 'name': 'Alice', 'age': 30}}
{'id': 3, 'data': {'type': 'UserDeleted', 'name': 'Alice'}}
Latest event:
{'id': 3, 'data': {'type': 'UserDeleted', 'name': 'Alice'}}
Fetched event #2:
{'id': 2, 'data': {'type': 'UserUpdated', 'name': 'Alice', 'age': 30}}
How it works
The EventStore class wraps a list _events that acts as an append-only log. Each append call computes the next sequential ID from the current list length, then stores a dict with both the ID and the raw event payload. get_events returns a shallow copy so callers can't mutate the internal list directly. get_event performs a simple linear scan over all events, which is fine for a mock or small in-memory store. The class simulates the core behavior of an event-sourcing system: you only append new facts, never update or delete history.
Common mistakes
- Returning the internal list directly from get_events, allowing external mutation of the store
- Forgetting that event IDs must be immutable — recomputing from list length can cause issues after future compaction or replay logic
- Assuming get_event returns by value forever; with a shallow copy you get the same dict refs, so mutations leak back
Variations
- Use a sqlite or Postgres table for durable storage with an AUTOINCREMENT ID column
- Read from a JSON file on startup and re-append to an in-memory list, simulating event sourcing replay
Real-world use cases
- Prototyping a microservice's event-sourcing layer before wiring up a real database.
- Testing command handlers in unit tests with an in-memory event store that supports quick rollback.
- Building a simple audit log for a legacy system without adding external infrastructure.
Sponsored
More from Microservices patterns
- BFF aggregation pattern: combine multiple service responses in Python easy
- Backward Compatible Schema Evolution in Python medium
- Bulkhead Thread Pool per Service Mock in Python medium
- CQRS with Separate Read and Write Repositories in Python medium
- Cache-Aside Pattern in Python: Per-Service Mock easy
- Consumer Driven Contract Pact Mock in Python medium
Keep learning
Related tutorials and quizzes for this topic.