Aggregate 3 - Explicit trigger and apply

This example shows another variation of the Dog aggregate class.

Like the previous example, this example uses the Aggregate class from the library. Event classes are defined explicitly to match command method signatures. In contrast to the previous example, this example explicitly triggers events within the command method bodies, and separately applies the events to the aggregate using an when() method decorated with singledispatchmethod with which event-specific methods are registered. To make this work, an Event class common to all the aggregate’s events is defined, which calls the aggregate’s apply() method from its apply() method.

Like in the previous examples, the application code simply uses the aggregate class as if it were a normal Python object class, albeit the aggregate class method register() is used to register a new dog.

Domain model

from typing import List, cast

from eventsourcing.dispatch import singledispatchmethod
from eventsourcing.domain import Aggregate

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

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

    class TrickAdded(Event):
        trick: str

    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)

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

    def _(self, event: Registered) -> None: =
        self.tricks: List[str] = []

    def _(self, event: TrickAdded) -> None:


from typing import Any, Dict
from uuid import UUID

from eventsourcing.application import Application
from eventsourcing.examples.aggregate3.domainmodel import Dog

class DogSchool(Application):
    is_snapshotting_enabled = True

    def register_dog(self, name: str) -> UUID:
        dog = Dog.register(name=name)

    def add_trick(self, dog_id: UUID, trick: str) -> None:
        dog: Dog = self.repository.get(dog_id)

    def get_dog(self, dog_id: UUID) -> Dict[str, Any]:
        dog: Dog = self.repository.get(dog_id)
        return {"name":, "tricks": tuple(dog.tricks)}

Test case

from unittest import TestCase

from eventsourcing.examples.aggregate3.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 =, 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")