Changed definition of apply() method in Aggregate base class.
There are difficulties of declaring type method between Python 3.7 and
3.10 without importing future annotations, but importing future annotations
somehow stops the dispatch register decorator working (says event is not
defined). Simply defining the method, and then overriding in subclass cures
this.
-from __future__ import annotations
-
-import sys
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime, timezone
@@ -50,10 +47,9 @@ class Aggregate:
self.apply(new_event)
self._pending_events.append(new_event)
- if sys.version_info >= (3, 8): # pragma: no cover
- apply: singledispatchmethod[None]
- else: # pragma: no cover
- apply: singledispatchmethod
+ @singledispatchmethod
+ def apply(self, event: DomainEvent) -> None:
+ """Applies event to aggregate."""
def collect_events(self) -> List[DomainEvent]:
events, self._pending_events = self._pending_events, []
