This is the first entry in a five-part series about event sourcing:
I hope this series has inspired you to try event sourcing. In this article, I will give you some suggestions on how to start using it.
The mechanics of event sourcing are not complicated. They’re different, and it takes some getting used to.
When I learned event sourcing, I was told to create my own code to implement the pattern. That has been a great learning experience, but there are so many decisions to factor in that I frequently felt stuck. Instead of following my path, I advise you to see how the pattern works with a framework that someone else wrote, and then feel free to adventure on your own.
I suggest you start by using the eventsourcing Python package on a few toy projects. By leveraging it, you can see how the patterns come together. John Bywater and team have done a lot of hard work to make it easy to hit the ground running when you decide to use event sourcing.
To introduce you to the framework, I’ll explore the main components of the package, especially since the package uses terminology that you might not be familiar with.
Understand the components
The main terms you’ll need to understand are Event, Aggregate, and Application. Let’s start with the Application.
Application
This is the outer shell of your event sourcing program. It exposes functionality to the outside world. APIs, web apps, or CLIs will call methods on your Application to perform tasks or retrieve data.
Applications in the eventsourcing package have a repository responsible for saving and retrieving events from the event store.
An incomplete application for a shopping cart could look like this:
from eventsourcing.application import Application
class CartApplication(Application):
def add_item(self, item_id, quantity, price, cart_id=None):
cart = self.repository.get(cart_id) if cart_id else Cart()
cart.add_item(item_id, quantity, price)
self.save(cart)
return cart.idIn domain-driven design terms, I think of the add_item method as a command. It represents the intention to change the system by adding an item to a cart. If given a cart_id , it’ll create a Cart object from the events in the event store. Otherwise it’ll create a new Cart . Then it will tell that object to add an item to itself, save the resulting events to the event store, and return the cart id.
I imagine this code looks somewhat similar to the code you’d write in a traditional app.
This is one thing I like about the eventsourcing package, it abstracts the mechanics away so your code can focus on the business logic.
But what is this Cart object? Let’s dive into that next.
Aggregate
The Cart object in the previous example is an aggregate. Aggregates are analogous to the term “model” in a traditional application, in that models represent some kind of entity that changes over time.
In traditional apps, creating a new Cart object would create a new row in a database with the values for that new object. Changing the Cart object would result in changing the same row in the database.
In an event-sourced app, creating an aggregate would create an event in the event store that represents the creating of the aggregate with the values for the new object. Changing the Cart object would result in adding a new event to the event store with an event representing that change.
This is a different approach, but your code probably won’t look much different.
An aggregate that represents a shopping cart can look like this:
from eventsourcing.domain import event, Aggregate
class Cart(Aggregate):
def __init__(self):
self._items = []
@event("ItemAdded")
def add_item(self, item_id, quantity, price):
self._items.append((item_id, quantity, price))The application object we discussed above would call the __init__ and add_item methods of this aggregate.
This code shows how well the eventsourcing package allows you to focus on your business logic. Calling one of these methods would result in creating events. Calling the __init__ method will create a Cart.Created event. Calling the add_item method would create a Cart.ItemAdded event.
In an event-sourced app, aggregate methods are responsible for changing the system and preventing situations that would clash with the business rules.
So, if the business wanted to prevent people from having more than five items in their cart, we would need to implement that in the add_item method. Maybe something like this:
class Cart(Aggregate):
...
@event("ItemAdded")
def add_item(self, item_id, quantity, price):
if len(self._items) >= 5:
raise MaxCartItemsException
self._items.append((item_id, quantity, price))As such, aggregates will hold the majority of your business logic.
Finally, let’s talk about events.
Event
An event represents a change that occurred in the system. Once created, it lives in the event store, usually in a database.
If I were to call the add_item method above and call repr() on the resulting event it could look like this:
Cart.ItemAdded(
originator_id=UUID('5e2c3aa2-86e6-4729-b751-13adc23d8da4'),
originator_version=2,
timestamp=datetime.datetime(2025, 3, 28, 1, 51, 32, 827273, tzinfo=datetime.timezone.utc),
item_id=324,
quantity=1,
price=3499
)This framework uses the term “originator” in its events to describe what other frameworks would call a model or entity. In this case, originator_id is akin to saying cart_id .
Events in the same event stream will have the same originator_id and a unique originator_version .
Another thing I like about the eventsourcing framework is that it does the work of defining the events for you from the parameters you provide in the aggregate’s methods. You can still define them yourself as well, to be more explicit, but I like that you have the option not to.
It’s your turn
This is enough to give you a quick overview of the package components. From here, I suggest you follow the package’s tutorial.
The tutorial spends most of its time working with in-memory “Plain Old Python Objects.” If you want to see the events in a database, you’ll have to configure the package to do so through environmental variables.