How to Build an Append-Only Event Store in Python
Implement a simple append-only event store class that stores events in a list and supports retrieval by index range.
Python code
30 linesclass EventStore:
def __init__(self):
self._events = []
def append(self, event):
"""Append an event to the store."""
self._events.append(event)
def get_events(self, start=0, end=None):
"""Return events from start index to end (exclusive)."""
return self._events[start:end]
def count(self):
"""Return the total number of events."""
return len(self._events)
if __name__ == "__main__":
store = EventStore()
store.append({"type": "UserCreated", "data": {"id": 1, "name": "Alice"}})
store.append({"type": "UserRenamed", "data": {"id": 1, "name": "Alicia"}})
store.append({"type": "UserDeleted", "data": {"id": 1}})
print("All events:")
for event in store.get_events():
print(f" {event['type']}: {event['data']}")
print(f"Total events: {store.count()}")
print("Events after index 1:")
print(store.get_events(1))
Output
All events:
UserCreated: {'id': 1, 'name': 'Alice'}
UserRenamed: {'id': 1, 'name': 'Alicia'}
UserDeleted: {'id': 1}
Total events: 3
Events after index 1:
[{'type': 'UserRenamed', 'data': {'id': 1, 'name': 'Alicia'}}, {'type': 'UserDeleted', 'data': {'id': 1}}]
How it works
The EventStore class wraps a Python list to provide two key operations: append to add an event and get_events to retrieve a slice of the sequence. The list internally ensures insertion order and provides O(1) append on average. The __init__ method initialises an empty list, and the get_events method uses Python's slice syntax to return a subset, which is safe because it returns a new list and does not mutate the original. The count method simply returns the length. This pattern enforces an append-only policy because there is no method to delete or modify existing events.
Common mistakes
- Using an external list directly instead of encapsulating it, which allows accidental mutation.
- Forgetting that slices create a new list, so large ranges can consume memory.
- Not defining a custom event ID or timestamp, making it hard to replay events in a distributed system.
Variations
- Use an SQLite database table with an AUTOINCREMENT primary key for persistence and concurrency.
- Use a deque or a custom iterable if you need efficient paging from the tail.
Real-world use cases
- Storing user action events in a microservice for audit logging and replaying them to rebuild materialised views.
- Recording order lifecycle events in an e-commerce platform to support compensation and analytics.
- Persisting financial transaction events in a banking application to maintain a tamper-evident transaction history.
Sponsored
More from System design patterns
- Build a BFF (Backend for Frontend) Mock Aggregator in Python medium
- Builder pattern for mocking complex objects in Python easy
- Circuit Breaker Pattern in Python: Closed, Open, and Half-Open States medium
- Create a Data Helper Class in Python easy
- Domain Driven Design Aggregate Root Example in Python medium
- Facade Pattern in Python with Mock Simplification medium
Keep learning
Related tutorials and quizzes for this topic.