Added EventSourcedLog (moved from Paxos/Replicated log example repo).
@@ -10,12 +10,14 @@ from typing import (
Dict,
Generic,
Iterable,
+ Iterator,
List,
Optional,
Sequence,
Type,
TypeVar,
Union,
+ cast,
)
from uuid import UUID
@@ -25,6 +27,7 @@ from eventsourcing.domain import (
DomainEvent,
Snapshot,
TAggregate,
+ TDomainEvent,
)
from eventsourcing.persistence import (
ApplicationRecorder,
@@ -750,3 +753,69 @@ class AggregateNotFound(Exception):
Raised when an :class:`~eventsourcing.domain.Aggregate`
object is not found in a :class:`Repository`.
"""
+
+
+class EventSourcedLog(Generic[TDomainEvent]):
+ """
+ Stored a sequence of domain events, like an aggregate.
+ But unlike an aggregate the events can be triggered
+ and selected for use in the application without
+ reconstructing a current state from all events.
+
+ This allows an indefinitely long sequence to be
+ generated, which is useful e.g. for logging and
+ discovering aggregate IDs of a particular type.
+ """
+
+ def __init__(
+ self,
+ events: EventStore[AggregateEvent[Aggregate]],
+ originator_id: UUID,
+ logged_cls: Type[TDomainEvent],
+ ):
+ self.events = events
+ self.originator_id = originator_id
+ self.logged_cls = logged_cls
+
+ def trigger_event(
+ self, next_originator_version: Optional[int] = None, **kwargs
+ ) -> TDomainEvent:
+ if next_originator_version is None:
+ last_logged = self.get_last()
+ if last_logged:
+ next_originator_version = last_logged.originator_version + 1
+ else:
+ next_originator_version = Aggregate.INITIAL_VERSION
+
+ return self.logged_cls(
+ originator_id=self.originator_id,
+ originator_version=next_originator_version,
+ timestamp=self.logged_cls.create_timestamp(),
+ **kwargs,
+ )
+
+ def get_last(self) -> Optional[TDomainEvent]:
+ # Get last logged event.
+ try:
+ return next(self.get(desc=True, limit=1))
+ except StopIteration:
+ return None
+
+ def get(
+ self,
+ gt: Optional[int] = None,
+ lte: Optional[int] = None,
+ desc: bool = False,
+ limit: Optional[int] = None,
+ ) -> Iterator[TDomainEvent]:
+ # Get logged events.
+ return cast(
+ Iterator[TDomainEvent],
+ self.events.get(
+ originator_id=self.originator_id,
+ gt=gt,
+ lte=lte,
+ desc=desc,
+ limit=limit,
+ ),
+ )
