Changed aggregate example 5 to use singledispatchmethod (was if/else).
@@ -3,6 +3,7 @@ from datetime import datetime, timezone
from typing import Any, Iterable, Optional, Tuple, Type, TypeVar
from uuid import UUID, uuid4
+from eventsourcing.dispatch import singledispatchmethod
from eventsourcing.domain import Snapshot
@@ -50,8 +51,9 @@ class Aggregate:
aggregate = cls.mutate(event, aggregate)
return aggregate
- @classmethod
- def mutate(cls, event: DomainEvent, aggregate: Any) -> Any:
+ @singledispatchmethod
+ @staticmethod
+ def mutate(event: DomainEvent, aggregate: Any) -> Any:
"""Mutates aggregate with event."""
@@ -68,9 +70,9 @@ class Dog(Aggregate):
class TrickAdded(DomainEvent):
trick: str
- @classmethod
- def register(cls, name: str) -> DomainEvent:
- return cls.Registered(
+ @staticmethod
+ def register(name: str) -> DomainEvent:
+ return Dog.Registered(
originator_id=uuid4(),
originator_version=1,
timestamp=DomainEvent.create_timestamp(),
@@ -78,12 +80,16 @@ class Dog(Aggregate):
)
def add_trick(self, trick: str) -> DomainEvent:
- return self.trigger_event(self.TrickAdded, trick=trick)
+ return self.trigger_event(Dog.TrickAdded, trick=trick)
+ @singledispatchmethod
@classmethod
def mutate(cls, event: DomainEvent, aggregate: Optional["Dog"]) -> Optional["Dog"]:
"""Mutates aggregate with event."""
- if isinstance(event, cls.Registered):
+
+ @mutate.register
+ @classmethod
+ def _(cls, event: Registered, _: Optional["Dog"]) -> "Dog":
return Dog(
id=event.originator_id,
version=event.originator_version,
@@ -92,7 +98,10 @@ class Dog(Aggregate):
name=event.name,
tricks=tuple(),
)
- elif isinstance(event, cls.TrickAdded):
+
+ @mutate.register
+ @classmethod
+ def _(cls, event: TrickAdded, aggregate: Optional["Dog"]) -> "Dog":
assert aggregate is not None
return Dog(
id=aggregate.id,
@@ -102,7 +111,10 @@ class Dog(Aggregate):
name=aggregate.name,
tricks=aggregate.tricks + (event.trick,),
)
- elif isinstance(event, Snapshot):
+
+ @mutate.register
+ @classmethod
+ def _(cls, event: Snapshot, _: Optional["Dog"]) -> "Dog":
return Dog(
id=event.state["id"],
version=event.state["version"],
@@ -111,5 +123,3 @@ class Dog(Aggregate):
name=event.state["name"],
tricks=tuple(event.state["tricks"]), # comes back from JSON as a list
)
- else:
- raise NotImplementedError(event) # pragma: no cover
