Tutorial - Part 1 - Getting Started¶
This tutorial provides a a very gradual introduction to event-sourced aggregates and applications, explaining just enough on the mechanics of this library to help users of the library understand how things work. It expands and explains the Synopsis, and prepares new users of the library for reading the Modules documentation.
Python classes¶
This tutorial depends on a basic understanding of Python classes.
For example, we can define a Dog class in Python as follows.
class Dog:
def __init__(self, name):
self.name = name
self.tricks = []
def add_trick(self, trick):
self.tricks.append(trick)
This example is taken from the Class and Instance Variables section of the Python docs.
Having defined a Python class, we can use it to create an instance.
dog = Dog('Fido')
The dog object is an instance of the Dog class.
assert isinstance(dog, Dog)
The __init__() method initialises the attributes name and tricks.
assert dog.name == 'Fido'
assert dog.tricks == []
The method add_trick() appends the argument trick to the attribute tricks.
dog.add_trick('roll over')
assert dog.tricks == ['roll over']
This is a simple example of a Python class.
In the next section, we convert the Dog class to be an event-sourced aggregate.
Event-sourced aggregate¶
A persistent object that changes through a sequence of decisions corresponds to the notion of an ‘aggregate’ in the book Domain-Driven Design. In the book, aggregates are persisted by inserting or updating database records that represent the current state of the object.
An event-sourced aggregate is persisted by recording the sequence of decisions as a sequence of ‘events’. This sequence of events is used to reconstruct the current state of the aggregate.
We can convert the Dog class into an event-sourced aggregate using
the Aggregate class and @event decorator from the library’s
domain module. Events will be triggered when
decorated methods are called. The changes are highlighted below.
from eventsourcing.domain import Aggregate, event
class Dog(Aggregate):
@event('Registered')
def __init__(self, name):
self.name = name
self.tricks = []
@event('TrickAdded')
def add_trick(self, trick):
self.tricks.append(trick)
As before, we can call the class to create a new instance.
dog = Dog('Fido')
The object is an instance of Dog. It is also an Aggregate.
assert isinstance(dog, Dog)
assert isinstance(dog, Aggregate)
As we might expect, the attributes name and tricks have been initialised.
assert dog.name == 'Fido'
assert dog.tricks == []
The dog aggregate also has an id attribute. The ID is used to uniquely identify
the aggregate within a collection of aggregates. It happens to be a UUID.
from uuid import UUID
assert isinstance(dog.id, UUID)
As above, we can call the method add_trick(). The given value is appended to tricks.
dog.add_trick('roll over')
assert dog.tricks == ['roll over']
By redefining the Dog class as an event-sourced aggregate in this way, we can
generate a sequence of event objects that can be used to reconstruct the aggregate.
We can get the events from the aggregate by calling collect_events().
events = dog.collect_events()
We can then reconstruct the aggregate by calling mutate() on the collected event objects.
copy = None
for e in events:
copy = e.mutate(copy)
assert copy == dog
Event-sourced aggregates can be developed and tested independently.
However, event-sourced aggregates are normally used within an application object, so that aggregate events can be stored in a database, and so that aggregates can be reconstructed from stored events.
Event-sourced application¶
Event-sourced applications combine event-sourced aggregates with a persistence mechanism to store and retrieve aggregate events.
We can define event-sourced applications with the Application class
from the library’s application module.
from eventsourcing.application import Application
We can save aggregates with the application’s save() method, and
reconstruct previously saved aggregates with the application repository’s
get() method.
Let’s define a DogSchool application that uses the Dog aggregate class.
class DogSchool(Application):
def register_dog(self, name):
dog = Dog(name)
self.save(dog)
return dog.id
def add_trick(self, dog_id, trick):
dog = self.repository.get(dog_id)
dog.add_trick(trick)
self.save(dog)
def get_dog(self, dog_id):
dog = self.repository.get(dog_id)
return {'name': dog.name, 'tricks': tuple(dog.tricks)}
The “command” methods register_dog() and add_trick() evolve application
state, and the “query” method get_dog() presents current state.
We can construct an instance of the application by calling the application class.
application = DogSchool()
We can then create and update aggregates by calling the command methods of the application.
dog_id = application.register_dog('Fido')
application.add_trick(dog_id, 'roll over')
application.add_trick(dog_id, 'fetch ball')
We can view the state of the aggregates by calling application query methods.
dog_details = application.get_dog(dog_id)
assert dog_details['name'] == 'Fido'
assert dog_details['tricks'] == ('roll over', 'fetch ball')
And we can propagate the state of the application as a whole by selecting event notifications from the application’s notification log.
notifications = application.notification_log.select(start=1, limit=10)
assert len(notifications) == 3
assert notifications[0].id == 1
assert notifications[1].id == 2
assert notifications[2].id == 3
Many different kinds of event-sourced applications can be defined in this way.
Project structure¶
You are free to structure your project files however you wish. You
may wish to put your aggregate classes in a file named
domainmodel.py and your application class in a file named
application.py.
myproject/
myproject/application.py
myproject/domainmodel.py
myproject/tests.py
Writing tests¶
It is generally recommended to follow a test-driven approach to the development of
event-sourced applications. You can get started with your event sourcing project by
first writing a failing test in tests.py. You can begin by defining your application
and aggregate classes in the test module. You can then refactor by moving aggregate and
application classes to separate Python modules. You can convert these modules to packages
if you want to split things up into smaller modules.
def test():
# Construct application object.
app = DogSchool()
# Call application command methods.
dog_id = app.register_dog('Fido')
app.add_trick(dog_id, 'roll over')
app.add_trick(dog_id, 'fetch ball')
# Call application query method.
assert app.get_dog(dog_id) == {
'name': 'Fido',
'tricks': (
'roll over',
'fetch ball'
)
}
Exercise¶
Try it for yourself by copying the code snippets above into your IDE, and running the test.
test()