Added support in the projections module for DCB applications.

Also some adjustments made whilst tracking down weird segmentation violation from projections with Python 3.13 and MsgStruct instances with zero attribute.

Also increased version to 9.5.0a3.

Author: johnbywater

docs/topics/projection.rst

@@ -27,6 +27,10 @@ Application subscriptions
 
 This module provides an :class:`~eventsourcing.projection.ApplicationSubscription` class, which can
 be used to "subscribe" to the domain events of an application.
+Please note, the :class:`~eventsourcing.popo.POPOApplicationRecorder` and
+:class:`~eventsourcing.postgres.PostgresApplicationRecorder` classes implement the
+required :func:`~eventsourcing.persistence.ApplicationRecorder.subscribe`
+method, but the :class:`~eventsourcing.sqlite.SQLiteApplicationRecorder` class does not.
 
 Application subscription objects are iterators that return domain events from an application sequence.
 Each domain event is accompanied by a tracking object that identifies the position of the
@@ -99,10 +103,48 @@ method which can be used to stop the subscription to the application recorder in
             subscription.stop()  # ...so we can continue with the examples
 
 
-Please note, the :class:`~eventsourcing.popo.POPOApplicationRecorder` and
-:class:`~eventsourcing.postgres.PostgresApplicationRecorder` classes implement the
-required :func:`~eventsourcing.persistence.ApplicationRecorder.subscribe`
-method, but the :class:`~eventsourcing.sqlite.SQLiteApplicationRecorder` class does not.
+The :class:`~eventsourcing.projection.DCBApplicationSubscription` class is the equivalent for
+:ref:`DCB applications <DCB application>`. It returns :ref:`tagged decisions <DCB Tagged>`.
+Please note, the :ref:`DCB recorders <DCB recorders>` :class:`~eventsourcing.dcb.popo.InMemoryDCBRecorder`,
+:class:`~eventsourcing.dcb.postgres_tt.PostgresDCBRecorderTT` and the ``eventsourcing_umadb`` extension
+all implement the required :func:`~eventsourcing.dcb.api.DCBRecorder.subscribe` method.
+
+.. code-block:: python
+
+    from uuid import UUID
+
+    from eventsourcing.dcb.application import DCBApplication
+    from eventsourcing.dcb.domain import Perspective, Tagged
+    from eventsourcing.dcb.msgpack import Decision, InitialDecision, MessagePackMapper
+    from eventsourcing.projection import DCBApplicationSubscription
+    from eventsourcing.utils import get_topic
+
+
+    # Define a perspective.
+    class MyPerspective(Perspective[Decision]):
+        @property
+        def cb(self) -> Selector | Sequence[Selector]:
+            return []
+
+
+    # Construct an application object.
+    app = DCBApplication(env={"MAPPER_TOPIC": get_topic(MessagePackMapper)})
+
+    # Record an event.
+    perspective = MyPerspective()
+    perspective.append_new_decision(
+        Tagged([], InitialDecision(originator_topic=""))
+    )
+    app.repository.save(perspective)
+
+    # Position in application sequence from which to subscribe.
+    max_tracking_id = 0
+
+    with DCBApplicationSubscription(app, gt=max_tracking_id, topics=()) as subscription:
+        for tagged_event, tracking in subscription:
+            # Process the event and record new state with tracking information.
+            subscription.stop()  # ...so we can continue with the examples
+
 
 .. _Projection:
 
@@ -150,7 +192,7 @@ The example below shows how a projection can be defined.
     from eventsourcing.projection import Projection
     from eventsourcing.utils import get_topic
 
-    class MyProjection(Projection["MyMaterialisedViewInterface"]):
+    class AggregateEventProjection(Projection["MyMaterialisedViewInterface"]):
         name = "myprojection"
         topics = (get_topic(Aggregate.Event), )
 
@@ -163,6 +205,35 @@ The example below shows how a projection can be defined.
             self.view.my_command(tracking)
 
 
+For projections that work with :ref:`DCB applications <DCB application>`, you will need to define the dispatching
+to work with :ref:`tagged decisions <DCB tagged>`. That is, because the ``process_event()`` method will receive
+:class:`~eventsourcing.dcb.domain.Tagged` objects, and because `singledispatchmethod` dispatches on the type of
+the first argument, you will need to forward ``tagged.decision`` and define handlers for different
+types of :class:`~eventsourcing.dcb.domain.Decision`.
+
+.. code-block:: python
+
+    from eventsourcing.dcb.domain import Tagged
+    from eventsourcing.dcb.msgpack import Decision, InitialDecision
+
+
+    class TaggedDecisionProjection(Projection["MyMaterialisedViewInterface"]):
+        name = "myprojection"
+        topics = (get_topic(Decision), )
+
+        @singledispatchmethod
+        def process_event(self, tagged: Tagged[Decision], tracking: Tracking) -> None:
+            self.process_decision(tagged.decision, tracking)
+
+        @singledispatchmethod
+        def process_decision(self, _: Decision, tracking: Tracking) -> None:
+            pass
+
+        @process_decision.register
+        def process_initial_decision(self, _: InitialDecision, tracking: Tracking) -> None:
+            self.view.my_command(tracking)
+
+
 The example below indicates how the projection's materialised view can be defined.
 
 .. code-block:: python
@@ -194,6 +265,7 @@ The example below indicates how the projection's materialised view can be define
         def my_command(self, tracking: Tracking) -> None:
             ...
 
+
 .. _Projection runner:
 
 Projection runner
@@ -249,7 +321,7 @@ signal after 1s.
     with ProjectionRunner(
         application_class=Application,
         view_class=MyPOPOMaterialisedView,
-        projection_class=MyProjection,
+        projection_class=AggregateEventProjection,
         env={},
     ) as projection_runner:
 
@@ -271,6 +343,37 @@ returned from calls to the event-sourced application's :func:`~eventsourcing.app
 method can be used by the user interface to :func:`~eventsourcing.persistence.TrackingRecorder.wait` until
 the "read model" has been updated.
 
+The projection runner supports :ref:`DCB application classes <DCB application>`.
+
+.. code-block:: python
+
+    import os, signal, threading, time
+
+    from eventsourcing.projection import ProjectionRunner
+
+    # For demonstration purposes, interrupt process with SIGINT after 1s.
+    def sleep_then_kill() -> None:
+        time.sleep(1)
+        os.kill(os.getpid(), signal.SIGINT)
+
+    threading.Thread(target=sleep_then_kill).start()
+
+    # Run projection as a context manager.
+    with ProjectionRunner(
+        application_class=DCBApplication,
+        view_class=MyPOPOMaterialisedView,
+        projection_class=TaggedDecisionProjection,
+        env={"MAPPER_TOPIC": get_topic(MessagePackMapper)},
+    ) as projection_runner:
+
+        # Register signal handler.
+        signal.signal(signal.SIGINT, lambda *args: projection_runner.stop())
+
+        # Run until interrupted.
+        projection_runner.run_forever()
+
+
+
 See :doc:`Tutorial - Part 4 </topics/tutorial/part4>` for more guidance and examples.
 
 
@@ -302,6 +405,8 @@ to increment a ``Counter`` aggregate.
 .. literalinclude:: ../../tests/projection_tests/test_event_sourced_projection.py
     :pyobject: Counter
 
+This library does does not yet support projections event-soured with :ref:`DCB applications <DCB application>`.
+
 
 .. _Event-sourced projection runner:
 
@@ -363,6 +468,8 @@ currently support application subscriptions.
         assert runner.projection.get_count(Aggregate.Event) == 1
 
 
+The event-sourced projection runner does not yet support projections event-soured with :ref:`DCB applications <DCB application>`.
+
 Code reference
 ==============
 

docs/topics/tutorial/part4.rst

@@ -205,13 +205,13 @@ is not capable of reconstructing.
 Example projection
 ==================
 
-For example, the ``EventCountersProjection`` class, shown below, processes events of an event-sourced application by
+For example, the ``AggregateEventCountersProjection`` class, shown below, processes events of an event-sourced application by
 calling methods of a concrete instance of ``EventCountersInterface``. It inherits the :class:`~eventsourcing.projection.Projection`
 class, setting the type argument as ``EventCountersInterface`` class. It sets the :py:attr:`~eventsourcing.projection.Projection.name`
 attribute as ``'eventcounters'``. It sets the :py:attr:`~eventsourcing.projection.Projection.topics` attribute
 to mention the topics of the domain events processed by this projection.
 
-The ``EventCountersProjection`` class implements the :func:`~eventsourcing.projection.Projection.process_event`
+The ``AggregateEventCountersProjection`` class implements the :func:`~eventsourcing.projection.Projection.process_event`
 by calling the ``incr_created_event_counter()`` and ``incr_subsequent_event_counter()`` methods of ``EventCountersInterface``
 available on :data:`~eventsourcing.projection.Projection.view`.
 
@@ -224,7 +224,7 @@ progress along an application sequence can be recorded, for example if large gap
 subscription returning checkpoints.
 
 .. literalinclude:: ../../../tests/projection_tests/test_projection.py
-    :pyobject: EventCountersProjection
+    :pyobject: AggregateEventCountersProjection
 
 
 Runners
@@ -272,11 +272,11 @@ to wait until an event has been processed by the projection before calling a que
 Counting events in memory
 =========================
 
-For example, the ``TestEventCountersProjection`` class, shown below, tests the ``EventCountersProjection``
+For example, the ``TestAggregateEventCountersProjection`` class, shown below, tests the ``AggregateEventCountersProjection``
 projection class with the ``POPOEventCounters`` class.
 
 .. literalinclude:: ../../../tests/projection_tests/test_projection.py
-    :pyobject: TestEventCountersProjection
+    :pyobject: TestAggregateEventCountersProjection
 
 The method ``test_event_counters_projection()`` constructs a runner, and from the runner gets references
 to an event-sourced application "write model" and a materialised view "read model".
@@ -286,7 +286,7 @@ instance of the application object and of the materialised view object as the pr
 for generating events and for getting the counted numbers of events.
 
 Events are generated in the event-sourced application "write model". The "created" and subsequent events are
-processed, by updating the materialised view "read model", according to the logic of the ``EventCountersProjection``
+processed, by updating the materialised view "read model", according to the logic of the ``AggregateEventCountersProjection``
 projection. The counted numbers of each kind of event are obtained from the "read model". The materialised view is
 "eventually consistent" because the event processing is asynchronous, and so the
 :func:`~eventsourcing.persistence.TrackingRecorder.wait` method is used to wait for the events to be
@@ -302,8 +302,8 @@ method will time out.
 Counting events in PostgreSQL
 =============================
 
-The ``TestEventCountersProjectionWithPostgres`` class extends ``TestEventCountersProjection`` and runs
-``EventCountersProjection`` with the ``PostgresEventCounters`` class.
+The ``TestAggregateEventCountersProjectionWithPostgres`` class extends ``TestAggregateEventCountersProjection`` and runs
+``AggregateEventCountersProjection`` with the ``PostgresEventCounters`` class.
 Because this test case uses a durable database for both the event-sourced application and the materialised view,
 any instance of the application can be used to write events, and any instance of the materialised view can be used to
 obtain the counted numbers of events. If a durable database is used in production, the event-sourced
@@ -318,7 +318,7 @@ caused by the ``SpannerThrown`` event. The event-sourced application and the mat
 databases. But in this example they are configured more simply to use different tables in the same database.
 
 .. literalinclude:: ../../../tests/projection_tests/test_projection.py
-    :pyobject: TestEventCountersProjectionWithPostgres
+    :pyobject: TestAggregateEventCountersProjectionWithPostgres
 
 
 Exercises

eventsourcing/dcb/api.py

@@ -3,7 +3,7 @@
 from abc import ABC, abstractmethod
 from collections.abc import Iterator
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Generic, TypeVar
 
 from eventsourcing.persistence import ProgrammingError
 
@@ -92,7 +92,7 @@ def subscribe(
         query: DCBQuery | None = None,
         *,
         after: int | None = None,
-    ) -> DCBSubscription:
+    ) -> DCBSubscription[Self]:
         """
         Returns all events, unless 'after' is given then only those with position
         greater than 'after', and unless any query items are given, then only those
@@ -103,10 +103,13 @@ def subscribe(
         """
 
 
-class DCBSubscription(Iterator[DCBSequencedEvent]):
+TDCBRecorder_co = TypeVar("TDCBRecorder_co", bound=DCBRecorder, covariant=True)
+
+
+class DCBSubscription(Iterator[DCBSequencedEvent], Generic[TDCBRecorder_co]):
     def __init__(
         self,
-        recorder: DCBRecorder,
+        recorder: TDCBRecorder_co,
         query: DCBQuery | None = None,
         after: int | None = None,
     ) -> None:

eventsourcing/dcb/application.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import os
-from typing import TYPE_CHECKING, Any, Generic
+from typing import TYPE_CHECKING, Any, Generic, cast
 
 from eventsourcing.dcb.domain import (
     EnduringObject,
@@ -15,6 +15,7 @@
 from eventsourcing.dcb.persistence import (
     DCBEventStore,
     DCBInfrastructureFactory,
+    DCBMapper,
     NotFoundError,
 )
 from eventsourcing.utils import Environment, EnvType, resolve_topic
@@ -37,10 +38,15 @@ def __init_subclass__(cls, **kwargs: Any) -> None:
     def __init__(self, env: EnvType | None = None):
         self.env = self.construct_env(self.name, env)
         self.factory = DCBInfrastructureFactory.construct(self.env)
-        self.recorder = self.factory.dcb_event_store()
+        self.recorder = self.factory.dcb_recorder()
         if "MAPPER_TOPIC" in self.env:
-            mapper = resolve_topic(self.env["MAPPER_TOPIC"])()
-            self.events = DCBEventStore(mapper, self.recorder)
+            # Only need a mapper, event store, and repository
+            # if we are using the higher-level abstractions.
+            self.mapper = cast(
+                DCBMapper[Any], resolve_topic(self.env["MAPPER_TOPIC"])()
+            )
+            assert isinstance(self.mapper, DCBMapper)
+            self.events = DCBEventStore(self.mapper, self.recorder)
             self.repository = DCBRepository(self.events)
 
     def construct_env(self, name: str, env: EnvType | None = None) -> Environment:
@@ -51,6 +57,9 @@ def construct_env(self, name: str, env: EnvType | None = None) -> Environment:
             _env.update(env)
         return Environment(name, _env)
 
+    def close(self) -> None:
+        self.factory.close()
+
     def __enter__(self) -> Self:
         self.factory.__enter__()
         return self
@@ -61,6 +70,7 @@ def __exit__(
         exc_val: BaseException | None,
         exc_tb: TracebackType | None,
     ) -> None:
+        self.close()
         self.factory.__exit__(exc_type, exc_val, exc_tb)
 
 

eventsourcing/dcb/persistence.py

@@ -15,6 +15,7 @@
     DCBRecorder,
     DCBSequencedEvent,
     DCBSubscription,
+    TDCBRecorder_co,
 )
 from eventsourcing.dcb.domain import (
     Selector,
@@ -49,6 +50,8 @@ def append(
         cb: Selector | Sequence[Selector] | None = None,
         after: int | None = None,
     ) -> int:
+        if len(events) == 0:
+            return 0
         if not cb and not after:
             condition = None
         else:
@@ -113,14 +116,14 @@ class NotFoundError(Exception):
 
 class DCBInfrastructureFactory(BaseInfrastructureFactory[TTrackingRecorder], ABC):
     @abstractmethod
-    def dcb_event_store(self) -> DCBRecorder:
+    def dcb_recorder(self) -> DCBRecorder:
         pass  # pragma: no cover
 
 
-class DCBListenNotifySubscription(DCBSubscription):
+class DCBListenNotifySubscription(DCBSubscription[TDCBRecorder_co]):
     def __init__(
         self,
-        recorder: DCBRecorder,
+        recorder: TDCBRecorder_co,
         query: DCBQuery | None = None,
         after: int | None = None,
     ) -> None: