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
@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: Registered) -> None:
self.name = event.name
self.tricks: List[str] = []
@apply.register
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.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¶
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")