Adjusted order of code in example aggregate example 6.
@@ -2,7 +2,7 @@ from dataclasses import dataclass
from datetime import datetime, timezone
from functools import reduce, singledispatch
from time import monotonic
-from typing import Callable, Iterable, List, Optional, Tuple, TypeVar, Union, cast
+from typing import Callable, Iterable, List, Optional, Tuple, Type, TypeVar, Union, cast
from uuid import UUID, uuid4
from eventsourcing.application import ProjectorFunctionType
@@ -16,6 +16,10 @@ class DomainEvent(HasOriginatorIDVersion):
timestamp: datetime
+def create_timestamp() -> datetime:
+ return datetime.fromtimestamp(monotonic(), timezone.utc)
+
+
@dataclass(frozen=True)
class Aggregate(HasIDVersionFields):
id: UUID
@@ -42,16 +46,22 @@ def aggregate_projector(
return project_aggregate
-def create_timestamp() -> datetime:
- return datetime.fromtimestamp(monotonic(), timezone.utc)
-
-
@dataclass(frozen=True)
class Dog(Aggregate):
name: str
tricks: Tuple[str, ...]
+@dataclass(frozen=True)
+class DogRegistered(DomainEvent):
+ name: str
+
+
+@dataclass(frozen=True)
+class TrickAdded(DomainEvent):
+ trick: str
+
+
def register_dog(name: str) -> Tuple[Dog, List[DomainEvent]]:
event = DogRegistered(
originator_id=uuid4(),
@@ -62,11 +72,6 @@ def register_dog(name: str) -> Tuple[Dog, List[DomainEvent]]:
return cast(Dog, mutate_dog(event, None)), [event]
-@dataclass(frozen=True)
-class DogRegistered(DomainEvent):
- name: str
-
-
def add_trick(dog: Dog, trick: str) -> Tuple[Dog, List[DomainEvent]]:
event = TrickAdded(
originator_id=dog.id,
@@ -77,11 +82,6 @@ def add_trick(dog: Dog, trick: str) -> Tuple[Dog, List[DomainEvent]]:
return cast(Dog, mutate_dog(event, dog)), [event]
-@dataclass(frozen=True)
-class TrickAdded(DomainEvent):
- trick: str
-
-
@singledispatch
def mutate_dog(
event: Union[DomainEvent, Snapshot[Dog]], dog: Optional[Dog]
@@ -90,7 +90,7 @@ def mutate_dog(
@mutate_dog.register
-def _(event: DogRegistered, _: Dog) -> Dog:
+def _(event: DogRegistered, _: Type[None]) -> Dog:
return Dog(
id=event.originator_id,
version=event.originator_version,
@@ -112,7 +112,7 @@ def _(event: TrickAdded, dog: Dog) -> Dog:
@mutate_dog.register(Snapshot)
-def _(event: Snapshot[Dog], _: Dog) -> Dog:
+def _(event: Snapshot[Dog], _: Type[None]) -> Dog:
return Dog(
id=event.state["id"],
version=event.state["version"],
