Updated to work with eventsourcing 9.4 (albeit not implemented separate tracking recorders yet, or subscriptions).

Author: johnbywater

Makefile

@@ -103,7 +103,11 @@ fmt: fmt-black fmt-isort
 
 .PHONY: test
 test:
-	$(POETRY) run python -m pytest $(opts) $(call tests,.)
+	$(POETRY) run python -m unittest discover . -v
+
+.PHONY: pytest
+pytest:
+	$(POETRY) run pytest . -v  --durations 10
 
 .PHONY: build
 build:

README.md

@@ -189,7 +189,7 @@ Define an adapter for a `scoped_session` object and configure the event-sourced
 application using the environment variable `SQLALCHEMY_SCOPED_SESSION_TOPIC`.
 
 ```python
-from eventsourcing.application import AggregateNotFound
+from eventsourcing.application import AggregateNotFoundError
 from eventsourcing.utils import get_topic
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker, scoped_session
@@ -242,7 +242,7 @@ session.remove()
 try:
     # forgot to commit
     app.repository.get(aggregate.id)
-except AggregateNotFound:
+except AggregateNotFoundError:
     pass
 else:
     raise Exception("Expected aggregate not found")
@@ -379,7 +379,7 @@ else:
 with db(commit_on_exit=True):
     try:
         es_app.repository.get(aggregate.id)
-    except AggregateNotFound:
+    except AggregateNotFoundError:
         pass
     else:
         raise Exception("Expected aggregate not found")

eventsourcing_sqlalchemy/__init__.py

@@ -1 +1 @@
-from .factory import Factory  # noqa
+from .factory import SQLAlchemyFactory  # noqa

eventsourcing_sqlalchemy/factory.py

@@ -1,11 +1,14 @@
 # -*- coding: utf-8 -*-
+from __future__ import annotations
+
 from typing import Optional
 
 from eventsourcing.persistence import (
     AggregateRecorder,
     ApplicationRecorder,
     InfrastructureFactory,
     ProcessRecorder,
+    TrackingRecorder,
 )
 from eventsourcing.utils import Environment, resolve_topic, strtobool
 from sqlalchemy.orm import scoped_session
@@ -18,7 +21,7 @@
 )
 
 
-class Factory(InfrastructureFactory):
+class SQLAlchemyFactory(InfrastructureFactory[TrackingRecorder]):
     SQLALCHEMY_URL = "SQLALCHEMY_URL"
     SQLALCHEMY_AUTOFLUSH = "SQLALCHEMY_AUTOFLUSH"
     SQLALCHEMY_CONNECTION_CREATOR_TOPIC = "SQLALCHEMY_CONNECTION_CREATOR_TOPIC"
@@ -102,6 +105,14 @@ def process_recorder(self) -> ProcessRecorder:
             recorder.create_table()
         return recorder
 
+    def tracking_recorder(
+        self, tracking_recorder_class: type[TrackingRecorder] | None = None
+    ) -> TrackingRecorder:
+        raise NotImplementedError
+
     def env_create_table(self) -> bool:
         default = "yes"
         return bool(strtobool(self.env.get(self.CREATE_TABLE) or default))
+
+
+Factory = SQLAlchemyFactory

eventsourcing_sqlalchemy/recorders.py

@@ -10,6 +10,7 @@
     Notification,
     ProcessRecorder,
     StoredEvent,
+    Subscription,
     Tracking,
 )
 from sqlalchemy import Column, Table, text
@@ -173,7 +174,7 @@ def insert_events(
                 notification_ids = self._insert_events(session, stored_events, **kwargs)
         return notification_ids
 
-    def max_notification_id(self) -> int:
+    def max_notification_id(self) -> int | None:
         try:
             with self.transaction(commit=False) as session:
                 # record_class = cast(Type[StoredEventRecord], self.events_record_cls)
@@ -183,20 +184,26 @@ def max_notification_id(self) -> int:
                 records = q[0:1]
                 return records[0].id
         except (IndexError, AssertionError):
-            return 0
+            return None
 
     def select_notifications(
         self,
-        start: int,
+        start: int | None,
         limit: int,
-        stop: Optional[int] = None,
+        stop: int | None = None,
         topics: Sequence[str] = (),
-    ) -> List[Notification]:
+        *,
+        inclusive_of_start: bool = True,
+    ) -> list[Notification]:
         with self.transaction(commit=False) as session:
             # record_class = cast(Type[StoredEventRecord], self.events_record_cls)
             record_class = self.events_record_cls
             q = session.query(record_class)
-            q = q.filter(record_class.id >= start)
+            if start is not None:
+                if inclusive_of_start:
+                    q = q.filter(record_class.id >= start)
+                else:
+                    q = q.filter(record_class.id > start)
             if stop is not None:
                 q = q.filter(record_class.id <= stop)
             if topics:
@@ -218,6 +225,12 @@ def select_notifications(
             ]
         return notifications
 
+    def subscribe(
+        self, gt: int | None = None, topics: Sequence[str] = ()
+    ) -> Subscription[ApplicationRecorder]:
+        msg = "SQLAlchemyApplicationRecorder.subscribe() is not implemented"
+        raise NotImplementedError(msg)
+
 
 class SQLAlchemyProcessRecorder(SQLAlchemyApplicationRecorder, ProcessRecorder):
     def __init__(
@@ -260,20 +273,27 @@ def _insert_events(
             session.add(record)
         return notification_ids
 
-    def max_tracking_id(self, application_name: str) -> int:
+    def max_tracking_id(self, application_name: str) -> int | None:
         with self.transaction(commit=False) as session:
             q = session.query(self.tracking_record_cls)
             q = q.filter(self.tracking_record_cls.application_name == application_name)
             q = q.order_by(self.tracking_record_cls.notification_id.desc())
             try:
                 max_id = q[0].notification_id
             except IndexError:
-                max_id = 0
+                max_id = None
         return max_id
 
-    def has_tracking_id(self, application_name: str, notification_id: int) -> bool:
+    def has_tracking_id(
+        self, application_name: str, notification_id: int | None
+    ) -> bool:
+        if notification_id is None:
+            return True
         with self.transaction(commit=False) as session:
             q = session.query(self.tracking_record_cls)
             q = q.filter(self.tracking_record_cls.application_name == application_name)
             q = q.filter(self.tracking_record_cls.notification_id == notification_id)
             return bool(q.count())
+
+    def insert_tracking(self, tracking: Tracking) -> None:
+        raise NotImplementedError

poetry.lock

@@ -29,7 +29,7 @@ python = "^3.9"
 #eventsourcing = { path = "../eventsourcing/", extras = ["crypto"], develop = true }
 #eventsourcing = { git = "https://github.com/pyeventsourcing/eventsourcing.git", branch = "main", extras = ["crypto"]}
 SQLAlchemy-Utils = ">=0.38.2"
-eventsourcing = "~9.3"
+eventsourcing = "~9.4"
 sqlalchemy = ">=1.4.26, <2.1"
 
 [tool.poetry.group.dev.dependencies]
@@ -45,7 +45,7 @@ isort = "*"
 mypy = "*"
 pre-commit = "*"
 pre-commit-hooks = "*"
-eventsourcing = { version = "~9.3", extras = ["crypto"] }
+eventsourcing = { version = "~9.4", extras = ["crypto"] }
 psycopg = { version = "*", extras = ["binary", "pool"] }
 psycopg2-binary = "*"
 pytest = "*"