Aggregate 2 - Explicit event classes

This example shows an aggregate that uses the library’s explicit syntax for defining aggregate events, as described in the tutorial and module docs. The difference between this example and example 1 is that the aggregate event classes are defined explicitly.

Domain model

The Dog class in this example uses the library’s aggregate base class and the event decorator. Event classes that match command method signatures are defined explicitly. Events are triggered by the decorator when the command methods are called, and the bodies of the command methods are used by the events to mutate the state of the aggregate, both after command methods are called and when reconstructing the state of an aggregate from stored events.

class Dog(Aggregate):
    class Registered(Aggregate.Created):
        name: str

    class TrickAdded(Aggregate.Event):
        trick: str

    @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

As in example 1, 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)
        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")

Code reference

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

Bases: Aggregate

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

Bases: Registered, Event

originator_id_type

alias of UUID

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

Bases: DecoratorEvent, TrickAdded

add_trick = <eventsourcing.domain.UnboundCommandMethodDecorator object>

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

Bases: Event, Created

originator_id_type

alias of UUID

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

Bases: Event

originator_id_type

alias of UUID

class examples.aggregate2.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.aggregate2.test_application.TestDogSchool(methodName='runTest')

Bases: TestCase

test_dog_school() → None