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 singledispatchmethod.

Like in the previous examples, the application code simply uses the aggregate class as if it were a normal Python object class.

Domain model

from __future__ import annotations

from typing import List, cast

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


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

    def __init__(self, name: str) -> None:
        self.name = name
        self.tricks: List[str] = []

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

    @singledispatchmethod
    def apply(self, event: Event) -> None:
        pass

    class TrickAdded(Aggregate.Event):
        trick: str

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

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

Application

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(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.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 = 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")