Aggregate 1 - Declarative syntax

This example shows an aggregate that uses the library’s declarative syntax for aggregates, as described in the tutorial and module docs.

Domain model

The Dog class in this example uses the library’s aggregate base class and the event decorator to define aggregate event classes from command method signatures. The event class names are given as the argument to the event decorator. The event attributes are defined automatically by the decorator to match the command method arguments. The bodies of the command methods are used to evolve the state of an aggregate instance, both when a new event is triggered and when an aggregate is reconstructed from stored events.

class Dog(Aggregate):

    @event("Registered")
    def __init__(self, name: str) -> None:
        self.name = name
        self.tricks: list[str] = []

    @event("TrickAdded")
    def add_trick(self, trick: str) -> None:
        self.tricks.append(trick)

Application

The DogSchool application class in this example uses the library’s application base class. It fully encapsulates the Dog aggregate, defining command and query methods that use the event-sourced aggregate class as if it were a normal Python object class.

class DogSchool(Application[UUID]):
    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

The TestDogSchool test case shows how the DogSchool application can be used.

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)
        self.assertEqual("Fido", dog["name"])
        self.assertEqual(("roll over", "play dead"), dog["tricks"])

        # Select notifications.
        notifications = school.notification_log.select(start=1, limit=10)
        self.assertEqual(3, len(notifications))

        # Take snapshot.
        school.take_snapshot(dog_id, version=3)
        dog = school.get_dog(dog_id)
        self.assertEqual("Fido", dog["name"])
        self.assertEqual(("roll over", "play dead"), dog["tricks"])

        # Continue with snapshotted aggregate.
        school.add_trick(dog_id, "fetch ball")
        dog = school.get_dog(dog_id)
        self.assertEqual("Fido", dog["name"])
        self.assertEqual(("roll over", "play dead", "fetch ball"), dog["tricks"])

Code reference

class examples.aggregate1.domainmodel.Dog(*args: Any, **kwargs: Any)

Bases: Aggregate

__init__(name: str) → None

add_trick = <eventsourcing.domain.UnboundCommandMethodDecorator object>

class Created(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: Event, Created

__init__(originator_id: UUID, originator_version: int, timestamp: datetime, originator_topic: str) → None

originator_id_type

alias of UUID

class Event(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: Event

__init__(originator_id: UUID, originator_version: int, timestamp: datetime) → None

originator_id_type

alias of UUID

class Registered(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', originator_topic: 'str')

Bases: Created, Event

__init__(originator_id: UUID, originator_version: int, timestamp: datetime, originator_topic: str, name: str) → None

originator_id_type

alias of UUID

name: str

class TrickAdded(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: DecoratorEvent, Event

__init__(originator_id: UUID, originator_version: int, timestamp: datetime, trick: str) → None

trick: str

class examples.aggregate1.application.DogSchool(env: Mapping[str, str] | None = None)

Bases: Application[UUID]

is_snapshotting_enabled: bool = True

register_dog(name: str) → UUID

add_trick(dog_id: UUID, trick: str) → None

get_dog(dog_id: UUID) → dict[str, Any]

name = 'DogSchool'

class examples.aggregate1.test_application.TestDogSchool(methodName='runTest')

Bases: TestCase

test_dog_school() → None