Extracted BaseInfrastructureFactory, so that "normal" and DCB factory classes can live in the same Python module, and also be detected and used separately.

Author: johnbywater

eventsourcing/dcb/persistence.py

@@ -17,7 +17,7 @@
     Tagged,
     TMutates,
 )
-from eventsourcing.persistence import InfrastructureFactory, TTrackingRecorder
+from eventsourcing.persistence import BaseInfrastructureFactory, TTrackingRecorder
 from eventsourcing.utils import get_topic
 
 if TYPE_CHECKING:
@@ -105,7 +105,7 @@ class NotFoundError(Exception):
     pass
 
 
-class DCBInfrastructureFactory(InfrastructureFactory[TTrackingRecorder], ABC):
+class DCBInfrastructureFactory(BaseInfrastructureFactory[TTrackingRecorder], ABC):
     @abstractmethod
     def dcb_event_store(self) -> DCBRecorder:
         pass  # pragma: no cover

eventsourcing/dcb/postgres_tt.py

@@ -17,8 +17,8 @@
 from eventsourcing.dcb.popo import SimpleDCBReadResponse
 from eventsourcing.persistence import IntegrityError, ProgrammingError
 from eventsourcing.postgres import (
+    BasePostgresFactory,
     PostgresDatastore,
-    PostgresFactory,
     PostgresRecorder,
     PostgresTrackingRecorder,
 )
@@ -601,7 +601,7 @@ class PsycopgDCBQueryItem(NamedTuple):
 
 
 class PostgresTTDCBFactory(
-    PostgresFactory,
+    BasePostgresFactory[PostgresTrackingRecorder],
     DCBInfrastructureFactory[PostgresTrackingRecorder],
 ):
     def dcb_event_store(self) -> DCBRecorder:

eventsourcing/persistence.py

@@ -663,18 +663,33 @@ class InfrastructureFactoryError(EventSourcingError):
     """Raised when an infrastructure factory cannot be created."""
 
 
-class InfrastructureFactory(ABC, Generic[TTrackingRecorder]):
+class BaseInfrastructureFactory(ABC, Generic[TTrackingRecorder]):
     """Abstract base class for infrastructure factories."""
 
     PERSISTENCE_MODULE = "PERSISTENCE_MODULE"
     TRANSCODER_TOPIC = "TRANSCODER_TOPIC"
-    MAPPER_TOPIC = "MAPPER_TOPIC"
     CIPHER_TOPIC = "CIPHER_TOPIC"
     COMPRESSOR_TOPIC = "COMPRESSOR_TOPIC"
-    IS_SNAPSHOTTING_ENABLED = "IS_SNAPSHOTTING_ENABLED"
-    APPLICATION_RECORDER_TOPIC = "APPLICATION_RECORDER_TOPIC"
-    TRACKING_RECORDER_TOPIC = "TRACKING_RECORDER_TOPIC"
-    PROCESS_RECORDER_TOPIC = "PROCESS_RECORDER_TOPIC"
+
+    def __init__(self, env: Environment | EnvType | None):
+        """Initialises infrastructure factory object with given application name."""
+        self.env = env if isinstance(env, Environment) else Environment(env=env)
+        self._is_entered = False
+
+    def __enter__(self) -> Self:
+        self._is_entered = True
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> None:
+        self._is_entered = False
+
+    def close(self) -> None:
+        """Closes any database connections, and anything else that needs closing."""
 
     @classmethod
     def construct(
@@ -747,11 +762,6 @@ def construct(
             raise InfrastructureFactoryError(msg)
         return factory_cls(env=env)
 
-    def __init__(self, env: Environment | EnvType | None):
-        """Initialises infrastructure factory object with given application name."""
-        self.env = env if isinstance(env, Environment) else Environment(env=env)
-        self._is_entered = False
-
     def transcoder(
         self,
     ) -> Transcoder:
@@ -763,32 +773,6 @@ def transcoder(
             transcoder_class = JSONTranscoder
         return transcoder_class()
 
-    def mapper(
-        self,
-        transcoder: Transcoder | None = None,
-        mapper_class: type[Mapper[TAggregateID]] | None = None,
-    ) -> Mapper[TAggregateID]:
-        """Constructs a mapper."""
-        # Resolve MAPPER_TOPIC if no given class.
-        if mapper_class is None:
-            mapper_topic = self.env.get(self.MAPPER_TOPIC)
-            mapper_class = (
-                resolve_topic(mapper_topic) if mapper_topic else Mapper[TAggregateID]
-            )
-
-        # Check we have a mapper class.
-        assert mapper_class is not None
-        origin_mapper_class = typing.get_origin(mapper_class) or mapper_class
-        assert isinstance(origin_mapper_class, type), mapper_class
-        assert issubclass(origin_mapper_class, Mapper), mapper_class
-
-        # Construct and return a mapper.
-        return mapper_class(
-            transcoder=transcoder or self.transcoder(),
-            cipher=self.cipher(),
-            compressor=self.compressor(),
-        )
-
     def cipher(self) -> Cipher | None:
         """Reads environment variables 'CIPHER_TOPIC'
         and 'CIPHER_KEY' to decide whether or not
@@ -822,6 +806,42 @@ def compressor(self) -> Compressor | None:
                 compressor = compressor_cls
         return compressor
 
+
+class InfrastructureFactory(BaseInfrastructureFactory[TTrackingRecorder]):
+    """Abstract base class for Application factories."""
+
+    MAPPER_TOPIC = "MAPPER_TOPIC"
+    IS_SNAPSHOTTING_ENABLED = "IS_SNAPSHOTTING_ENABLED"
+    APPLICATION_RECORDER_TOPIC = "APPLICATION_RECORDER_TOPIC"
+    TRACKING_RECORDER_TOPIC = "TRACKING_RECORDER_TOPIC"
+    PROCESS_RECORDER_TOPIC = "PROCESS_RECORDER_TOPIC"
+
+    def mapper(
+        self,
+        transcoder: Transcoder | None = None,
+        mapper_class: type[Mapper[TAggregateID]] | None = None,
+    ) -> Mapper[TAggregateID]:
+        """Constructs a mapper."""
+        # Resolve MAPPER_TOPIC if no given class.
+        if mapper_class is None:
+            mapper_topic = self.env.get(self.MAPPER_TOPIC)
+            mapper_class = (
+                resolve_topic(mapper_topic) if mapper_topic else Mapper[TAggregateID]
+            )
+
+        # Check we have a mapper class.
+        assert mapper_class is not None
+        origin_mapper_class = typing.get_origin(mapper_class) or mapper_class
+        assert isinstance(origin_mapper_class, type), mapper_class
+        assert issubclass(origin_mapper_class, Mapper), mapper_class
+
+        # Construct and return a mapper.
+        return mapper_class(
+            transcoder=transcoder or self.transcoder(),
+            cipher=self.cipher(),
+            compressor=self.compressor(),
+        )
+
     def event_store(
         self,
         mapper: Mapper[TAggregateID] | None = None,
@@ -858,21 +878,6 @@ def is_snapshotting_enabled(self) -> bool:
         """
         return strtobool(self.env.get(self.IS_SNAPSHOTTING_ENABLED, "no"))
 
-    def __enter__(self) -> Self:
-        self._is_entered = True
-        return self
-
-    def __exit__(
-        self,
-        exc_type: type[BaseException] | None,
-        exc_val: BaseException | None,
-        exc_tb: TracebackType | None,
-    ) -> None:
-        self._is_entered = False
-
-    def close(self) -> None:
-        """Closes any database connections, and anything else that needs closing."""
-
 
 @dataclass(frozen=True)
 class Tracking:

eventsourcing/postgres.py

@@ -28,6 +28,7 @@
 from eventsourcing.persistence import (
     AggregateRecorder,
     ApplicationRecorder,
+    BaseInfrastructureFactory,
     DatabaseError,
     DataError,
     InfrastructureFactory,
@@ -45,6 +46,7 @@
     Subscription,
     Tracking,
     TrackingRecorder,
+    TTrackingRecorder,
 )
 from eventsourcing.utils import Environment, EnvType, resolve_topic, retry, strtobool
 
@@ -1109,7 +1111,7 @@ def _insert_events(
         super()._insert_events(curs, stored_events, **kwargs)
 
 
-class PostgresFactory(InfrastructureFactory[PostgresTrackingRecorder]):
+class BasePostgresFactory(BaseInfrastructureFactory[TTrackingRecorder]):
     POSTGRES_DBNAME = "POSTGRES_DBNAME"
     POSTGRES_HOST = "POSTGRES_HOST"
     POSTGRES_PORT = "POSTGRES_PORT"
@@ -1132,11 +1134,6 @@ class PostgresFactory(InfrastructureFactory[PostgresTrackingRecorder]):
     POSTGRES_ENABLE_DB_FUNCTIONS = "POSTGRES_ENABLE_DB_FUNCTIONS"
     CREATE_TABLE = "CREATE_TABLE"
 
-    aggregate_recorder_class = PostgresAggregateRecorder
-    application_recorder_class = PostgresApplicationRecorder
-    tracking_recorder_class = PostgresTrackingRecorder
-    process_recorder_class = PostgresProcessRecorder
-
     def __init__(self, env: Environment | EnvType | None):
         super().__init__(env)
         dbname = self.env.get(self.POSTGRES_DBNAME)
@@ -1334,6 +1331,32 @@ def __init__(self, env: Environment | EnvType | None):
     def env_create_table(self) -> bool:
         return strtobool(self.env.get(self.CREATE_TABLE) or "yes")
 
+    def __enter__(self) -> Self:
+        self.datastore.__enter__()
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> None:
+        self.datastore.__exit__(exc_type, exc_val, exc_tb)
+
+    def close(self) -> None:
+        with contextlib.suppress(AttributeError):
+            self.datastore.close()
+
+
+class PostgresFactory(
+    BasePostgresFactory[PostgresTrackingRecorder],
+    InfrastructureFactory[PostgresTrackingRecorder],
+):
+    aggregate_recorder_class = PostgresAggregateRecorder
+    application_recorder_class = PostgresApplicationRecorder
+    tracking_recorder_class = PostgresTrackingRecorder
+    process_recorder_class = PostgresProcessRecorder
+
     def aggregate_recorder(self, purpose: str = "events") -> AggregateRecorder:
         prefix = self.env.name.lower() or "stored"
         events_table_name = prefix + "_" + purpose
@@ -1412,21 +1435,5 @@ def process_recorder(self) -> ProcessRecorder:
             recorder.create_table()
         return recorder
 
-    def __enter__(self) -> Self:
-        self.datastore.__enter__()
-        return self
-
-    def __exit__(
-        self,
-        exc_type: type[BaseException] | None,
-        exc_val: BaseException | None,
-        exc_tb: TracebackType | None,
-    ) -> None:
-        self.datastore.__exit__(exc_type, exc_val, exc_tb)
-
-    def close(self) -> None:
-        with contextlib.suppress(AttributeError):
-            self.datastore.close()
-
 
 Factory = PostgresFactory

eventsourcing/tests/persistence.py

@@ -629,7 +629,7 @@ def insert_events() -> None:
         rate = num_jobs * num_events_per_job / (ended - started).total_seconds()
         print(f"Rate: {rate:.0f} inserts per second")
 
-    def optional_test_insert_subscribe(self) -> None:
+    def optional_test_insert_subscribe(self, initial_position: int = 0) -> None:
 
         recorder = self.create_recorder()
 
@@ -655,7 +655,9 @@ def optional_test_insert_subscribe(self) -> None:
 
         notification_ids = recorder.insert_events([stored_event1, stored_event2])
         if self.EXPECT_CONTIGUOUS_NOTIFICATION_IDS:
-            self.assertEqual(notification_ids, [1, 2])
+            self.assertEqual(
+                notification_ids, [1 + initial_position, 2 + initial_position]
+            )
 
         # Get the max notification ID.
         max_notification_id2 = recorder.max_notification_id()
@@ -698,8 +700,8 @@ def optional_test_insert_subscribe(self) -> None:
                 stored_event2.originator_version, notifications[1].originator_version
             )
             if self.EXPECT_CONTIGUOUS_NOTIFICATION_IDS:
-                self.assertEqual(1, notifications[0].id)
-                self.assertEqual(2, notifications[1].id)
+                self.assertEqual(1 + initial_position, notifications[0].id)
+                self.assertEqual(2 + initial_position, notifications[1].id)
 
             # Store a third event.
             stored_event3 = StoredEvent(
@@ -710,7 +712,7 @@ def optional_test_insert_subscribe(self) -> None:
             )
             notification_ids = recorder.insert_events([stored_event3])
             if self.EXPECT_CONTIGUOUS_NOTIFICATION_IDS:
-                self.assertEqual(notification_ids, [3])
+                self.assertEqual(notification_ids, [3 + initial_position])
 
             # Receive events from the subscription.
             for notification in subscription:
@@ -726,7 +728,7 @@ def optional_test_insert_subscribe(self) -> None:
                 stored_event3.originator_version, notifications[2].originator_version
             )
             if self.EXPECT_CONTIGUOUS_NOTIFICATION_IDS:
-                self.assertEqual(3, notifications[2].id)
+                self.assertEqual(3 + initial_position, notifications[2].id)
 
         # Start a subscription with int value for 'start'.
         with recorder.subscribe(gt=max_notification_id2) as subscription:
@@ -744,7 +746,7 @@ def optional_test_insert_subscribe(self) -> None:
             )
 
         # Start a subscription, call stop() during iteration.
-        with recorder.subscribe(gt=None) as subscription:
+        with recorder.subscribe(gt=initial_position) as subscription:
 
             # Receive events from the subscription.
             for i, _ in enumerate(subscription):

examples/coursebookingdcb/postgres_ts.py

@@ -17,8 +17,8 @@
 from eventsourcing.dcb.popo import SimpleDCBReadResponse
 from eventsourcing.persistence import IntegrityError, ProgrammingError
 from eventsourcing.postgres import (
+    BasePostgresFactory,
     PostgresDatastore,
-    PostgresFactory,
     PostgresRecorder,
     PostgresTrackingRecorder,
 )
@@ -447,7 +447,7 @@ class PgDCBEventRow(TypedDict):
 
 
 class PostgresTSDCBFactory(
-    PostgresFactory,
+    BasePostgresFactory[PostgresTrackingRecorder],
     DCBInfrastructureFactory[PostgresTrackingRecorder],
 ):
     def dcb_event_store(self) -> DCBRecorder: