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 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
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")