Aggregate 2 - Explicit event classes¶
This example shows a slightly different version of the Dog
class used in the
tutorial and module docs.
It uses the Aggregate
class and the @event
decorator from the library,
but explicitly defines event classes to match command method signatures. As in
the previous example, the event are triggered when the command methods are called,
and the bodies of the command methods are used to apply the events to the aggregate
state.
As in the previous example, the application code simply uses the aggregate class as if it were a normal Python object class.
Domain model¶
from typing import List
from eventsourcing.domain import Aggregate, event
class Dog(Aggregate):
class Registered(Aggregate.Created):
name: str
@event(Registered)
def __init__(self, name: str) -> None:
self.name = name
self.tricks: List[str] = []
class TrickAdded(Aggregate.Event):
trick: str
@event(TrickAdded)
def add_trick(self, trick: str) -> None:
self.tricks.append(trick)
Application¶
from typing import Any, Dict
from uuid import UUID
from eventsourcing.application import Application
from eventsourcing.examples.aggregate2.domainmodel import Dog
class DogSchool(Application):
is_snapshotting_enabled = True
def register_dog(self, name: str) -> UUID:
dog = Dog(name)
self.save(dog)
return dog.id
def add_trick(self, dog_id: UUID, trick: str) -> None:
dog: Dog = self.repository.get(dog_id)
dog.add_trick(trick)
self.save(dog)
def get_dog(self, dog_id: UUID) -> Dict[str, Any]:
dog: Dog = self.repository.get(dog_id)
return {"name": dog.name, "tricks": tuple(dog.tricks)}
Test case¶
from unittest import TestCase
from eventsourcing.examples.aggregate2.application import DogSchool
class TestDogSchool(TestCase):
def test_dog_school(self) -> None:
# Construct application object.
school = DogSchool()
# Evolve application state.
dog_id = school.register_dog("Fido")
school.add_trick(dog_id, "roll over")
school.add_trick(dog_id, "play dead")
# Query application state.
dog = school.get_dog(dog_id)
assert dog["name"] == "Fido"
assert dog["tricks"] == ("roll over", "play dead")
# Select notifications.
notifications = school.notification_log.select(start=1, limit=10)
assert len(notifications) == 3
# Take snapshot.
school.take_snapshot(dog_id, version=3)
dog = school.get_dog(dog_id)
assert dog["name"] == "Fido"
assert dog["tricks"] == ("roll over", "play dead")
# Continue with snapshotted aggregate.
school.add_trick(dog_id, "fetch ball")
dog = school.get_dog(dog_id)
assert dog["name"] == "Fido"
assert dog["tricks"] == ("roll over", "play dead", "fetch ball")