diff --git a/eventsourcing/examples/aggregate6a/domainmodel.py b/eventsourcing/examples/aggregate6a/domainmodel.py index 626320fb..7e1068ce 100644 --- a/eventsourcing/examples/aggregate6a/domainmodel.py +++ b/eventsourcing/examples/aggregate6a/domainmodel.py @@ -29,6 +29,19 @@ class Aggregate: created_on: datetime modified_on: datetime + def hold_event(self, event: DomainEvent) -> None: + all_pending_events[id(self)].append(event) + + def collect_events(self) -> List[DomainEvent]: + try: + return all_pending_events.pop(id(self)) + except KeyError: # pragma: no cover + return [] + + def __del__(self) -> None: + with contextlib.suppress(KeyError): + all_pending_events.pop(id(self)) + TAggregate = TypeVar("TAggregate", bound=Aggregate) MutatorFunction = Callable[..., Optional[TAggregate]] @@ -55,19 +68,6 @@ class Dog(Aggregate): name: str tricks: Tuple[str, ...] - def __del__(self) -> None: - with contextlib.suppress(KeyError): - all_pending_events.pop(id(self)) - - def hold_event(self, event: DomainEvent) -> None: - all_pending_events[id(self)].append(event) - - def collect_events(self) -> List[DomainEvent]: - try: - return all_pending_events.pop(id(self)) - except KeyError: # pragma: no cover - return [] - @dataclass(frozen=True) class DogRegistered(DomainEvent): diff --git a/eventsourcing/examples/aggregate7a/domainmodel.py b/eventsourcing/examples/aggregate7a/domainmodel.py index 6504bfbe..4cea933a 100644 --- a/eventsourcing/examples/aggregate7a/domainmodel.py +++ b/eventsourcing/examples/aggregate7a/domainmodel.py @@ -34,6 +34,19 @@ class Aggregate(BaseModel): class Config: frozen = True + def hold_event(self, event: DomainEvent) -> None: + all_pending_events[id(self)].append(event) + + def collect_events(self) -> List[DomainEvent]: + try: + return all_pending_events.pop(id(self)) + except KeyError: # pragma: no cover + return [] + + def __del__(self) -> None: + with contextlib.suppress(KeyError): + all_pending_events.pop(id(self)) + class Snapshot(DomainEvent): topic: str @@ -78,19 +91,6 @@ class Dog(Aggregate): name: str tricks: Tuple[Trick, ...] - def __del__(self) -> None: - with contextlib.suppress(KeyError): - all_pending_events.pop(id(self)) - - def hold_event(self, event: DomainEvent) -> None: - all_pending_events[id(self)].append(event) - - def collect_events(self) -> List[DomainEvent]: - try: - return all_pending_events.pop(id(self)) - except KeyError: # pragma: no cover - return [] - class DogRegistered(DomainEvent): name: str