Aggregate 3 - Explicit trigger and apply

This example shows how to explicitly trigger events within aggregate command methods, and an alternative style for implementing aggregate projector functions.

Domain model

The Dog class in this example uses the library’s aggregate base class. However, this example does not use the event decorator that was used in example 1 and example 2, but instead explicitly triggers aggregate events from within command method bodies, by calling trigger_event.

It also defines a separate aggregate projector function, apply() which is decorated with @singledispatchmethod. Event-specific methods are registered with the apply() method, and invoked when the method is called with that type of event. To make this work, an Event class common to all the aggregate’s events is defined, which calls the aggregate’s apply() method.

class Dog(Aggregate):
    class Event(Aggregate.Event):
        def apply(self, aggregate: Dog) -> None:
            aggregate.apply(self)

    class Registered(Event, Aggregate.Created):
        name: str

    class TrickAdded(Event):
        trick: str

    @classmethod
    def register(cls, name: str) -> Dog:
        return cls._create(cls.Registered, name=name)

    def add_trick(self, trick: str) -> None:
        self.trigger_event(self.TrickAdded, trick=trick)

    @singledispatchmethod
    def apply(self, event: Event) -> None:
        """Applies event to aggregate."""

    @apply.register
    def _(self, event: Dog.Registered) -> None:
        self.name = event.name
        self.tricks: list[str] = []

    @apply.register
    def _(self, event: Dog.TrickAdded) -> None:
        self.tricks.append(event.trick)

Application

As in example 1 and example 2, 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.register(name=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.aggregate3.domainmodel.Dog(*args: Any, **kwargs: Any)

Bases: Aggregate

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

Bases: Event

apply(aggregate: Dog) → None

Applies the domain event to its aggregate.

This method does nothing but exist to be overridden as a convenient way for users to define how an event evolves the state of an aggregate.

originator_id_type

alias of UUID

timestamp: datetime

Timezone-aware datetime object representing when an event occurred.

originator_id: TAggregateID

UUID identifying an aggregate to which the event belongs.

originator_version: int

Integer identifying the version of the aggregate when the event occurred.

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

Bases: Event, Created

name: str

originator_id_type

alias of UUID

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

Bases: Event

trick: str

originator_id_type

alias of UUID

classmethod register(name: str) → Dog

Register a virtual subclass of an ABC.

Returns the subclass, to allow usage as a class decorator.

add_trick(trick: str) → None

apply(event: Event) → None

apply(event: Registered) → None

apply(event: TrickAdded) → None

Applies event to aggregate.

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

Bases: Event, Created

originator_id_type

alias of UUID

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

Bases: TestCase

test_dog_school() → None