Aggregate 3 - Separate mutator¶
This example shows the Dog
class used in the tutorial and module docs.
Like the previous example, it uses the Aggregate
class from the library,
and defines event classes and that are defined explicitly to match command
method signatures. In contrast from the previous example, it 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")