Source code for eventsourcing.application

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, Generic, List, Optional, TypeVar
from uuid import UUID

from eventsourcing.domain import (
    Aggregate,
    AggregateEvent,
    Snapshot,
    TAggregate,
)
from eventsourcing.persistence import (
    ApplicationRecorder,
    DatetimeAsISO,
    DecimalAsStr,
    EventStore,
    InfrastructureFactory,
    JSONTranscoder,
    Mapper,
    Notification,
    Transcoder,
    UUIDAsHex,
)


[docs]class Repository(Generic[TAggregate]): """Reconstructs aggregates from events in an :class:`~eventsourcing.persistence.EventStore`, possibly using snapshot store to avoid replaying all events."""
[docs] def __init__( self, event_store: EventStore[AggregateEvent], snapshot_store: Optional[EventStore[Snapshot]] = None, ): """ Initialises repository with given event store (an :class:`~eventsourcing.persistence.EventStore` for aggregate :class:`~eventsourcing.domain.AggregateEvent` objects) and optionally a snapshot store (an :class:`~eventsourcing.persistence.EventStore` for aggregate :class:`~eventsourcing.domain.Snapshot` objects). """ self.event_store = event_store self.snapshot_store = snapshot_store
[docs] def get(self, aggregate_id: UUID, version: Optional[int] = None) -> TAggregate: """ Returns an :class:`~eventsourcing.domain.Aggregate` for given ID, optionally at the given version. """ aggregate: Optional[TAggregate] = None gt: Optional[int] = None if self.snapshot_store is not None: # Try to get a snapshot. snapshots = self.snapshot_store.get( originator_id=aggregate_id, desc=True, limit=1, lte=version, ) try: snapshot = next(snapshots) except StopIteration: pass else: gt = snapshot.originator_version aggregate = snapshot.mutate() # Get aggregate events. domain_events = self.event_store.get( originator_id=aggregate_id, gt=gt, lte=version, ) # Reconstruct the aggregate from its events. for domain_event in domain_events: aggregate = domain_event.mutate(aggregate) # Raise exception if "not found". if aggregate is None: raise AggregateNotFound((aggregate_id, version)) else: # Return the aggregate. return aggregate
[docs]@dataclass(frozen=True) class Section: # noinspection PyUnresolvedReferences """ Frozen dataclass that represents a section from a :class:`NotificationLog`. The :data:`items` attribute contains a list of :class:`~eventsourcing.persistence.Notification` objects. The :data:`id` attribute is the section ID, two integers separated by a comma that described the first and last notification ID that are included in the section. The :data:`next_id` attribute describes the section ID of the next section, and will be set if the section contains as many notifications are were requested. Constructor arguments: :param Optional[str] id: section ID of this section e.g. "1,10" :param List[Notification] items: a list of event notifications :param Optional[str] next_id: section ID of the following section """ id: Optional[str] items: List[Notification] next_id: Optional[str]
[docs]class NotificationLog(ABC): """ Abstract base class for notification logs. """
[docs] @abstractmethod def __getitem__(self, section_id: str) -> Section: """ Returns a :class:`Section` from a notification log. """
[docs]class LocalNotificationLog(NotificationLog): """ Notification log that presents sections of event notifications retrieved from an :class:`~eventsourcing.persistence.ApplicationRecorder`. """ DEFAULT_SECTION_SIZE = 10
[docs] def __init__( self, recorder: ApplicationRecorder, section_size: int = DEFAULT_SECTION_SIZE, ): """ Initialises a local notification object with given :class:`~eventsourcing.persistence.ApplicationRecorder` and an optional section size. Constructor arguments: :param ApplicationRecorder recorder: application recorder from which event notifications will be selected :param int section_size: number of notifications to include in a section """ self.recorder = recorder self.section_size = section_size
[docs] def __getitem__(self, requested_section_id: str) -> Section: """ Returns a :class:`Section` of event notifications based on the requested section ID. The section ID of the returned section will describe the event notifications that are actually contained in the returned section, and may vary from the requested section ID if there are less notifications in the recorder than were requested, or if there are gaps in the sequence of recorded event notification. """ # Interpret the section ID. parts = requested_section_id.split(",") part1 = int(parts[0]) part2 = int(parts[1]) start = max(1, part1) limit = min(max(0, part2 - start + 1), self.section_size) # Select notifications. notifications = self.recorder.select_notifications(start, limit) # Get next section ID. actual_section_id: Optional[str] next_id: Optional[str] if len(notifications): last_notification_id = notifications[-1].id actual_section_id = self.format_section_id( notifications[0].id, last_notification_id ) if len(notifications) == limit: next_id = self.format_section_id( last_notification_id + 1, last_notification_id + limit ) else: next_id = None else: actual_section_id = None next_id = None # Return a section of the notification log. return Section( id=actual_section_id, items=notifications, next_id=next_id, )
@staticmethod def format_section_id(first_id: int, last_id: int) -> str: return "{},{}".format(first_id, last_id)
[docs]class Application(ABC, Generic[TAggregate]): """ Base class for event-sourced applications. """
[docs] def __init__(self) -> None: """ Initialises an application with an :class:`~eventsourcing.persistence.InfrastructureFactory`, a :class:`~eventsourcing.persistence.Mapper`, an :class:`~eventsourcing.persistence.ApplicationRecorder`, an :class:`~eventsourcing.persistence.EventStore`, a :class:`~eventsourcing.application.Repository`, and a :class:`~eventsourcing.application.LocalNotificationLog`. """ self.factory = self.construct_factory() self.mapper = self.construct_mapper() self.recorder = self.construct_recorder() self.events = self.construct_event_store() self.snapshots = self.construct_snapshot_store() self.repository: Repository[TAggregate] = self.construct_repository() self.log = self.construct_notification_log()
[docs] def construct_factory(self) -> InfrastructureFactory: """ Constructs an :class:`~eventsourcing.persistence.InfrastructureFactory` for use by the application. """ return InfrastructureFactory.construct(self.__class__.__name__)
[docs] def construct_mapper(self, application_name: str = "") -> Mapper: """ Constructs a :class:`~eventsourcing.persistence.Mapper` for use by the application. """ return self.factory.mapper( transcoder=self.construct_transcoder(), application_name=application_name, )
[docs] def construct_transcoder(self) -> Transcoder: """ Constructs a :class:`~eventsourcing.persistence.Transcoder` for use by the application. """ transcoder = JSONTranscoder() self.register_transcodings(transcoder) return transcoder
# noinspection SpellCheckingInspection
[docs] def register_transcodings(self, transcoder: Transcoder) -> None: """ Registers :class:`~eventsourcing.persistence.Transcoding` objects on given :class:`~eventsourcing.persistence.JSONTranscoder`. """ transcoder.register(UUIDAsHex()) transcoder.register(DecimalAsStr()) transcoder.register(DatetimeAsISO())
[docs] def construct_recorder(self) -> ApplicationRecorder: """ Constructs an :class:`~eventsourcing.persistence.ApplicationRecorder` for use by the application. """ return self.factory.application_recorder()
[docs] def construct_event_store(self) -> EventStore[AggregateEvent]: """ Constructs an :class:`~eventsourcing.persistence.EventStore` for use by the application to store and retrieve aggregate :class:`~eventsourcing.domain.AggregateEvent` objects. """ return self.factory.event_store( mapper=self.mapper, recorder=self.recorder, )
[docs] def construct_snapshot_store(self) -> Optional[EventStore[Snapshot]]: """ Constructs an :class:`~eventsourcing.persistence.EventStore` for use by the application to store and retrieve aggregate :class:`~eventsourcing.domain.Snapshot` objects. """ if not self.factory.is_snapshotting_enabled(): return None recorder = self.factory.aggregate_recorder(purpose="snapshots") return self.factory.event_store( mapper=self.mapper, recorder=recorder, )
[docs] def construct_repository(self) -> Repository[TAggregate]: """ Constructs a :class:`Repository` for use by the application. """ return Repository( event_store=self.events, snapshot_store=self.snapshots, )
[docs] def construct_notification_log(self) -> LocalNotificationLog: """ Constructs a :class:`LocalNotificationLog` for use by the application. """ return LocalNotificationLog(self.recorder, section_size=10)
[docs] def save(self, *aggregates: Aggregate, **kwargs: Any) -> None: """ Collects pending events from given aggregates and puts them in the application's event store. """ events = [] for aggregate in aggregates: events += aggregate.collect_events() self.events.put(events, **kwargs) self.notify(events)
[docs] def notify(self, new_events: List[AggregateEvent]) -> None: """ Called after new domain events have been saved. This method on this class class doesn't actually do anything, but this method may be implemented by subclasses that need to take action when new domain events have been saved. """
[docs] def take_snapshot(self, aggregate_id: UUID, version: Optional[int] = None) -> None: """ Takes a snapshot of the recorded state of the aggregate, and puts the snapshot in the snapshot store. """ if self.snapshots is None: raise AssertionError( "Can't take snapshot without snapshots store. Please " "set environment variable IS_SNAPSHOTTING_ENABLED to " "a true value (e.g. 'y')." ) else: aggregate = self.repository.get(aggregate_id, version) snapshot = Snapshot.take(aggregate) self.snapshots.put([snapshot])
TApplication = TypeVar("TApplication", bound=Application)
[docs]class AggregateNotFound(Exception): """ Raised when an :class:`~eventsourcing.domain.Aggregate` object is not found in a :class:`Repository`. """