Event Sourcing in Python: Get More Insights Into Your Data

Chris May knew about event sourcing for over a decade before he used it for real. Once he tried it in production, he was sold. After hearing him describe the pattern, I can see why, so in this article I'll walk you through what event sourcing is and why it changes what questions your data can answer.

The Problem: CRUD loses data

To make this tangible imagine two users completing an exercise on the Rust platform. With a CRUD (Create, Read, Update, Delete) approach, you would have a table that records the final state of each user's progress:

User starts → submits once → passes.

However, this only records the end state. What actually happens in the exercise lifecycle for a user is much richer:

User starts → submits (fails) → submits (fails) → submits (fails) → requests hint → submits (passes).

In a CRUD app, both rows look identical:

What you lost:

• How many attempts did each user need?
• Which exercises cause the most failures?
• Do users who request hints complete faster or slower?
• What is the conversion path from free to paying?

There might even be questions you do not know you have yet. With just CRUD you overwrite data, which you need to make up for with ETL pipelines, bolted on analytics, and guesswork. With event sourcing, you have all the data you need to answer these questions immediately. It just requires a bit more planning upfront, designing the right events and commands.

# What CRUD sees: identical end states
state_a.total_score == state_b.total_score == 10  # True, indistinguishable

# What event sourcing sees:
state_a.attempts["ownership"] == 1   # sailed through
state_b.attempts["ownership"] == 4   # struggled
state_b.hints_used["ownership"] == 1 # needed help

(This is from the test suite of a demo app we'll look at later; the full comparison test is test_crud_loses_data_event_sourcing_keeps_it.)

What Event Sourcing Actually Is

Essentially with event sourcing you have two types of tables: the event table and the projection tables. The event table is your source of truth: an append-only log of everything that has happened. The projection tables are optimized read models, built by listeners that fire when events arrive.

Appends are cheap (like Python list.append), while updates and deletes are not; shifting to append-only means you pay less and keep more. You derive the current state by replaying the events. This also takes advantage of storage being much cheaper than compute.

I also like the immutability aspect: events are facts, not mutable state, similar to having to explicitly say mut for mutable variables in Rust.

There are no updates and no deletes. A "delete" is just an event: ExerciseCancelled, CartClosed. State is derived by replaying events, never stored directly.

This has its origins in DDD (Domain-Driven Design); Greg Young coined the term "event sourcing", though the concept is much older: accounting ledgers, shipping logs, double-entry bookkeeping. Software just needed a name for it.

Chris described it well in our conversation:

You just keep writing events to a table. The complexity shifts from mutations to queries, which is a better problem to have.

Here is what the events look like in our demo app, a coding platform where users complete exercises. Each event captures a specific action that happened, along with relevant data and a timestamp.

@dataclass(frozen=True)
class ExerciseStarted:
    stream_id: str
    exercise: str
    timestamp: datetime = field(default_factory=_now)

@dataclass(frozen=True)
class TestsFailed:
    stream_id: str
    exercise: str
    output: str          # the exact compiler error: preserved forever
    timestamp: datetime = field(default_factory=_now)

frozen means immutable; once created, you cannot change the event.

Rebuilding state from events

As mentioned, state is not stored, it's computed by replaying the event stream. This is a fundamental shift from CRUD, where you store the current state directly. You get an audit trail for free and can replay to any point in time, powerful for debugging and for questions you haven't thought to ask yet.

Here is rebuild_state, which takes the event stream and returns the current state:

def rebuild_state(events: list[Event]) -> ExerciseState:
    state = ExerciseState()
    for event in events:
        match event:
            case ExerciseStarted(exercise=ex):
                state.exercises_in_progress.add(ex)
            case CodeSubmitted(exercise=ex):
                state.attempts[ex] = state.attempts.get(ex, 0) + 1
            case HintRequested(exercise=ex):
                state.hints_used[ex] = state.hints_used.get(ex, 0) + 1
            case ExerciseCompleted(exercise=ex, points=pts):
                state.exercises_in_progress.discard(ex)
                state.exercises_completed[ex] = pts
                state.total_score += pts
    return state

And here we see another paradigm I really like: functional programming. The rebuild_state function is a pure function; same input, same output, no side effects. This makes it easy to test and debug. Replay the same events, get the same state every time.

Commands: validate before you produce events

Events are one half. Commands are the other: requests to change state, validated against the current state before producing events. This is where your business rules live, similar to how I covered FSM transition validation.

Chris's example: "no more than one item added per minute"; with event sourcing, you load the stream, check the timestamps, and reject or accept based on that history. With CRUD, you would need a separate audit table. When your boss asks for 30-day performance metrics, the structured history is already there for AI tools to query.

Here is how execute_command validates a start request:

def execute_command(store: EventStore, *, stream_id: str, command: str, **kwargs):
    events = store.load_stream(stream_id)
    state = rebuild_state(events)      # current reality

    match command:
        case "start_exercise":
            exercise = kwargs["exercise"]
            if exercise in state.exercises_in_progress:
                raise CommandError(f"{exercise} already in progress")
            if exercise in state.exercises_completed:
                raise CommandError(f"{exercise} already completed")
            event = ExerciseStarted(stream_id=stream_id, exercise=exercise)
        ...

    store.append(event)
    return event

Business rule enforcement is just: load stream → rebuild state → check conditions → append or reject.

CQRS: Separating Reads from Writes

Next up is CQRS, which stands for Command Query Responsibility Segregation, a different pattern from event sourcing, but they pair naturally.

Chris told a story that made this click for me:

"Every now and then we would have to create a report for important people in the organization. Whenever they clicked the button — 'okay, I want the report now' — it would take so many resources to build that report in memory that it slowed down the rest of the application. Users could tell when they did it because all of a sudden the application just slowed down. We didn't use event sourcing at the time, but what we figured out we could do is just update the report on the server in place. And so when the executive said 'okay, I want that report,' it was now a static file download instead of building the whole thing up."

That is the projection pattern in a nutshell: the expensive work happens on write, so the read is instant.

In our coding platform example we can build two projections from the same event stream: a leaderboard that shows the top scores, and a progress view that shows which exercises a user has completed and which are in progress. Both projections listen to the same events, but they build different read models optimized for their specific queries.

def build_leaderboard(store: EventStore) -> list[tuple[str, int]]:
    scores = Counter()
    for event in store.all_events():
        if isinstance(event, ExerciseCompleted):
            scores[event.stream_id] += event.points
    return scores.most_common()

def build_progress(store: EventStore, stream_id: str) -> UserProgress:
    events = store.load_stream(stream_id)
    state = rebuild_state(events)
    return UserProgress(
        completed=set(state.exercises_completed),
        in_progress=state.exercises_in_progress,
    )

Vertical slice architecture pairs naturally with event sourcing. Every bit of code related to one feature lives in one folder. Independent. No cross-cutting dependencies, no "which file handles this again?"

Chris had a dashboard showing 200 items. A user asked for filtering: query by this, sort by that. His first instinct: "This is really going to complicate the code." Then he realized he could just create a new slice. Duplicate the folder, add the filtering logic, run it alongside the original. Iterate with the user until it was right, then swap it in. The old version kept running the whole time.

Rust's strong safety guarantees come to mind here. With Rust, you have to be explicit about mutability and ownership. This forces you to think about data flow and side effects in a way that is similar to event sourcing. You are always aware of when you are changing state and when you are not. (Rust has that effect on how you think about code.)

There is an AI angle here too. When Chris asks Claude Code to change something in the event sourced code, it reads one folder and produces good results. When it touches the CRUD code, it has to scan the whole codebase to figure out what is involved. This is slower, more token greedy, and more error prone. The vertical slice architecture with event sourcing is much more AI-friendly because it has clear boundaries and less coupling.

Gradual adoption works the same way. You do not convert your whole app. Pick one component, define events for it, start a new append table alongside your existing CRUD tables, and try it there. Chris was hesitant for a long time, partly because he could not see how event sourcing could coexist with regular CRUD code. Now he has seen that the two different approaches can actually co-exist.

A practical demo

"The only way to learn a new programming language is by writing programs in it" (Dennis Ritchie). Same goes for understanding a new design pattern.

The demo below walks you through building an event sourcing system from scratch, test by test, using the coding platform domain example from this article. You will start with a blank slate and end up with a fully functional event sourcing implementation, complete with commands, projections, and a comparison to CRUD.

The demo has six levels, each unlocked by making a group of tests pass:

git clone git@github.com:bbelderbos/event_sourcing_demo.git
cd event_sourcing_demo
uv sync
cp event_sourcing_template.py event_sourcing.py
uv run pytest test_event_sourcing.py -v
# work through the levels, making the tests pass

Work through event_sourcing_template.py top to bottom. Each step has a hint. The tests tell you exactly when you are done.

Where to go from here

Start with my conversation with Chris that inspired this article: Chris May on Event Sourcing, Simpler Frontends & AI Workflows.

Two more resources Chris recommends:

Package: eventsourcing by John Bywater: over a decade of active maintenance. Uses base classes and decorators to hide the ceremony. The Event sourcing in 15 minutes video in the docs, starts from scratch with TDD, installing the package, etc.; very approachable.

Book: Understanding Event Sourcing by Martin Dilger. Written in Java but with minimal code, easy to follow, but profound.

Some patterns are old and do not expire. They just wait for new use cases. I hope you find a use case for this pattern soon, and that Chris and I have helped you get started.

Actually, as I was publishing this article, Chris pushed up the first version of his intro event sourcing ebook, check it out here, as well as his latest Talk Python interview.

Keep reading

• What Rust Structs Taught Me About State and Ownership
• AI Agent Architecture in Python
• The Repository Pattern: Swappable Data Sources