Some month ago I got interested in CQRS and event sourcing approach. Yesterday, I took time to experiment with event sourcing a bit with Python.
The goal of event sourcing approach is to induce changes into a system by saving all the events that appends in it rather than mutating its state. Here is some possiblities that it brings:
- you do not have to know the wholes state of the entities of your system. All you need is their unique identifier to create events related to them,
- all you have to care at first is the persistence of your event storem
- you can theorically bring back the system to any state of its history, take a snapshot and represent it the way you want.
- as you store functional events (create user, sell item, whatever), data migration is not needed. All you have to think of is the way you propaged the event through your whole information system.
So the idea is pretty cool. It seems disconcerting at first. Though, notice that that's the way work our well known relational data bases management systems. Every transaction details is at first written in redo logs, even if it is in progress and not committed. This way, the RDBMS is able to rerun a set of operation in case of a crash, reducing the risk of losing an important part of a transaction. This is event sourcing! And it is consider as the safest way to track all the transactions in the system.
Let's look at some code. The stuff I wrote in the train is about to create people entities, or in other word associate them with an identifier in the system and being able to change their situation, like marital status or tell they have moved. It could be useful for a revenue declaration app.
I begin with creating a service to deal with people. I call it PersonRegistry. With it, I can create a person in the system and change her marital status:
class PersonRegistry:
"""Provide services to deal with people"""
# timeline parameter is my event store
def __init__(self, timeline):
"""Initialize timeline event"""
self.timeline = timeline
self._currentId = 0
def create(self, person):
"""Create a person in the system"""
self._currentId += 1
self.timeline.addEvent({
'type': EventTimeLine.PERSON_CREATION,
'personId': self._currentId,
'status': person.status,
'address': person.address.to_dict(),
'name': person.name.to_dict()
})
return self._currentId
def changeStatus(self, personId, newStatus):
self.timeline.addEvent({
'type': EventTimeLine.PERSON_STATUS_CHANGE,
'personId': personId,
'newStatus': newStatus
})
I wanted the way to store events to be very strait forward. They're simple dicts, easily serializable and readable.
To act as a serious Domain Driven Design guy, I created some value objects as well:
class Person:
SINGLE = 1
MARRIED = 2
def __init__(self, status, address, name):
self.status = status
self.address = address
self.name = name
class Name:
def __init__(self, firstname, lastname):
"""Blah Blah Blah"""
self.firstname, self.lastname = firstname, lastname
def to_dict(self):
return {'firstname': self.firstname, 'lastname': self.lastname}
class Address:
def __init__(self, street, city):
self.street = street
self.city = city
def to_dict(self):
return {'street': self.street, 'city': self.city}
The event store is called EventTimeLine. What it does is allow to addEvent in
the system and retrieve them by iterating in its data.
class EventTimeLine:
"""Basically, the list of all events in the system. Allows to add an event.
The object itself is iterable if you want to browse the created events."""
PERSON_CREATION = 1
PERSON_STATUS_CHANGE = 2
def __init__(self):
"""Initialize the inner event list"""
self.event_list = []
def addEvent(self, eventData):
"""Add an event in the event list."""
eventData['_datetime'] = datetime.datetime.today()
self.event_list.append(eventData)
def __iter__(self):
for e in self.event_list:
yield e
I can store my events. It is a good start for the first iteration. I can ship it to the users, so thay can populating their system with people.
Next step is to be able to get a snapshot of the system state. I'll be my monday iteration! For now, you can find the code on github.
