Beyond State Overwriting: Implementing Event Sourcing in Python

Most applications function by overwriting state. When a player picks up a sword in a game, a database record changes from two to three. This approach is efficient but destructive; it discards the history of how that state was reached. Event Sourcing flips this paradigm. Instead of storing the final balance or the current inventory count, you store a sequence of immutable events—facts that have happened in the past.

By replaying these events from the beginning, you can reconstruct the state at any point in time. This provides an inherent audit log, simplifies debugging by allowing "time travel," and enables the creation of multiple "projections" or views of the same data without altering the source of truth. It is the same fundamental logic that powers Git and Blockchain.

Prerequisites

To follow this implementation, you should have a solid grasp of Python 3.10+ fundamentals, specifically Object-Oriented Programming (OOP). Familiarity with dataclasses and enum is essential, as these provide the structure for immutable events. A basic understanding of Dependency Injection will also help when connecting the inventory logic to the underlying event store.

Key Libraries & Tools

• enum: Used to define distinct, readable event types like ITEM_ADDED and ITEM_REMOVED.
• dataclasses: Provides a concise way to create event objects, specifically using frozen=True to ensure immutability.
• collections: A specialized dictionary subclass for counting hashable objects, used here to aggregate inventory totals.
• functools: Implements a simple memoization strategy to avoid replaying the entire event history on every read request.
• Flox: A tool for creating reproducible development environments, ensuring consistent package management across different machines.

Code Walkthrough: Building the Core System

Step 1: Defining Immutable Events

We start by defining what an event looks like. It must contain the type of action and the data associated with it.

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto

class EventType(Enum):
    ITEM_ADDED = auto()
    ITEM_REMOVED = auto()

@dataclass(frozen=True)
class Event:
    type: EventType
    data: str
    timestamp: datetime = field(default_factory=datetime.now)

The frozen=True parameter is vital. Events represent the past; they cannot be changed once they occur. We use a default_factory for the timestamp to ensure each event is accurately placed in the timeline.

Step 2: The Event Store and Caching

The Event Store is a simple append-only list. To prevent performance degradation as the list grows, we apply a cache to the state reconstruction method.

from functools import cache
from collections import Counter

class Inventory:
    def __init__(self, store):
        self.store = store

    @cache
    def get_items(self):
        counts = Counter()
        for event in self.store.get_all_events():
            if event.type == EventType.ITEM_ADDED:
                counts[event.data] += 1
            elif event.type == EventType.ITEM_REMOVED:
                counts[event.data] -= 1
        return {k: v for k, v in counts.items() if v > 0}

    def _invalidate_cache(self):
        self.get_items.cache_clear()

When we add an item, we append an event to the store and trigger _invalidate_cache(). The next time get_items() is called, it recalculates and recaches the state.

Step 3: Advanced Projections

Projections allow us to ask different questions of our data. For example, we can determine which items were collected most frequently, regardless of whether they are still in the inventory.

def get_most_collected(store):
    events = store.get_all_events()
    added_items = [e.data for e in events if e.type == EventType.ITEM_ADDED]
    return Counter(added_items).most_common(3)

Syntax Notes

This implementation relies on Generic Type Variables (TypeVar) when evolving the system to handle complex objects rather than just strings. Using typing.Generic[T] allows the Event and EventStore classes to remain flexible, supporting any data structure while maintaining type safety. The use of the decorator pattern via @cache demonstrates a clean way to separate performance concerns from business logic.

Practical Examples

• Financial Systems: Storing every transaction (credit/debit) instead of just the balance to provide a perfect audit trail.
• E-commerce: Tracking how long items sit in a cart before being removed to analyze user hesitation.
• Gaming: Building a replay system by storing player inputs as events to recreate the match exactly.

Tips & Gotchas

• Schema Evolution: If you change the structure of your Item object later, your old events might break. You must plan for "upcasting" (transforming old events into the new format) or versioning your event schemas.
• Snapshotting: For systems with millions of events, replaying from zero is too slow even with local caching. Periodically save a "snapshot" of the state so you only have to replay events from the last snapshot forward.
• Avoid for CRUD: If your application only requires basic create, read, update, and delete operations without any need for history, event sourcing will introduce unnecessary complexity.