ISSUE #99

* Refactor `SagaExecutionStorage` as `SagaExecutionRepository`.
* Add `DatabaseSagaExecutionRepository`.

Author: Sergio García Prado

packages/core/minos-microservice-saga/minos/saga/__init__.py

@@ -46,6 +46,7 @@
 )
 from .executions import (
     ConditionalSagaStepExecution,
+    DatabaseSagaExecutionRepository,
     Executor,
     LocalExecutor,
     LocalSagaStepExecution,
@@ -54,7 +55,7 @@
     ResponseExecutor,
     SagaExecution,
     SagaExecutionDatabaseOperationFactory,
-    SagaExecutionStorage,
+    SagaExecutionRepository,
     SagaStatus,
     SagaStepExecution,
     SagaStepStatus,

packages/core/minos-microservice-saga/minos/saga/executions/__init__.py

@@ -1,15 +1,17 @@
 from .commit import (
     TransactionCommitter,
 )
-from .database import (
-    SagaExecutionDatabaseOperationFactory,
-)
 from .executors import (
     Executor,
     LocalExecutor,
     RequestExecutor,
     ResponseExecutor,
 )
+from .repositories import (
+    DatabaseSagaExecutionRepository,
+    SagaExecutionDatabaseOperationFactory,
+    SagaExecutionRepository,
+)
 from .saga import (
     SagaExecution,
 )
@@ -23,6 +25,3 @@
     RemoteSagaStepExecution,
     SagaStepExecution,
 )
-from .storage import (
-    SagaExecutionStorage,
-)

packages/core/minos-microservice-saga/minos/saga/executions/repositories/__init__.py

@@ -0,0 +1,7 @@
+from .abc import (
+    SagaExecutionRepository,
+)
+from .database import (
+    DatabaseSagaExecutionRepository,
+    SagaExecutionDatabaseOperationFactory,
+)

packages/core/minos-microservice-saga/minos/saga/executions/repositories/abc.py

@@ -0,0 +1,62 @@
+from abc import (
+    ABC,
+    abstractmethod,
+)
+from typing import (
+    Union,
+)
+from uuid import (
+    UUID,
+)
+
+from minos.common import (
+    SetupMixin,
+)
+
+from ..saga import (
+    SagaExecution,
+)
+
+
+class SagaExecutionRepository(SetupMixin, ABC):
+    """TODO"""
+
+    async def store(self, execution: SagaExecution) -> None:
+        """Store an execution.
+
+        :param execution: Execution to be stored.
+        :return: This method does not return anything.
+        """
+        return await self._store(execution)
+
+    async def _store(self, execution: SagaExecution) -> None:
+        raise NotImplementedError
+
+    async def load(self, uuid: Union[str, UUID]) -> SagaExecution:
+        """Load the saga execution stored on the given key.
+
+        :param uuid: The key to identify the execution.
+        :return: A ``SagaExecution`` instance.
+        """
+        if not isinstance(uuid, UUID):
+            uuid = UUID(uuid)
+        return await self._load(uuid)
+
+    @abstractmethod
+    async def _load(self, uuid: UUID) -> SagaExecution:
+        raise NotImplementedError
+
+    async def delete(self, uuid: Union[SagaExecution, str, UUID]) -> None:
+        """Delete the reference of the given key.
+
+        :param uuid: Execution key to be deleted.
+        :return: This method does not return anything.
+        """
+        if isinstance(uuid, SagaExecution):
+            uuid = uuid.uuid
+        if not isinstance(uuid, UUID):
+            uuid = UUID(uuid)
+        return await self._delete(uuid)
+
+    async def _delete(self, key: UUID) -> None:
+        raise NotImplementedError

packages/core/minos-microservice-saga/minos/saga/executions/database/__init__.py renamed to packages/core/minos-microservice-saga/minos/saga/executions/repositories/database/__init__.py

@@ -1,3 +1,6 @@
 from .factories import (
     SagaExecutionDatabaseOperationFactory,
 )
+from .impl import (
+    DatabaseSagaExecutionRepository,
+)

packages/core/minos-microservice-saga/minos/saga/executions/database/factories.py renamed to packages/core/minos-microservice-saga/minos/saga/executions/repositories/database/factories.py

@@ -0,0 +1,68 @@
+from __future__ import (
+    annotations,
+)
+
+from contextlib import (
+    suppress,
+)
+from typing import (
+    Type,
+)
+from uuid import (
+    UUID,
+)
+
+from minos.common import (
+    Config,
+    MinosConfigException,
+    MinosJsonBinaryProtocol,
+    MinosStorage,
+    MinosStorageLmdb,
+)
+
+from ....exceptions import (
+    SagaExecutionNotFoundException,
+)
+from ...saga import (
+    SagaExecution,
+)
+from ..abc import (
+    SagaExecutionRepository,
+)
+
+
+class DatabaseSagaExecutionRepository(SagaExecutionRepository):
+    """Saga Execution Storage class."""
+
+    def __init__(
+        self,
+        storage_cls: Type[MinosStorage] = MinosStorageLmdb,
+        protocol=MinosJsonBinaryProtocol,
+        db_name: str = "LocalState",
+        *args,
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+        self.db_name = db_name
+        self._storage = storage_cls.build(protocol=protocol, **kwargs)
+
+    @classmethod
+    def _from_config(cls, config: Config, **kwargs) -> SagaExecutionRepository:
+        with suppress(MinosConfigException):
+            kwargs |= config.get_database_by_name("saga")
+        return cls(**kwargs)
+
+    async def _store(self, execution: SagaExecution) -> None:
+        key = str(execution.uuid)
+        value = execution.raw
+        self._storage.update(table=self.db_name, key=key, value=value)
+
+    async def _delete(self, uuid: UUID) -> None:
+        self._storage.delete(table=self.db_name, key=str(uuid))
+
+    async def _load(self, uuid: UUID) -> SagaExecution:
+        value = self._storage.get(table=self.db_name, key=str(uuid))
+        if value is None:
+            raise SagaExecutionNotFoundException(f"The execution identified by {uuid} was not found.")
+        execution = SagaExecution.from_raw(value)
+        return execution

packages/core/minos-microservice-saga/minos/saga/executions/repositories/database/impl.py

@@ -45,8 +45,9 @@
     SagaPausedExecutionStepException,
 )
 from .executions import (
+    DatabaseSagaExecutionRepository,
     SagaExecution,
-    SagaExecutionStorage,
+    SagaExecutionRepository,
     SagaStatus,
 )
 from .messages import (
@@ -66,7 +67,7 @@ class SagaManager(SetupMixin):
     @Inject()
     def __init__(
         self,
-        storage: SagaExecutionStorage,
+        storage: SagaExecutionRepository,
         broker_pool: Optional[BrokerClientPool] = None,
         pool_factory: Optional[PoolFactory] = None,
         *args,
@@ -92,9 +93,17 @@ def _from_config(cls, *args, config: Config, **kwargs) -> SagaManager:
         :param kwargs: Additional named arguments.
         :return: A new ``SagaManager`` instance.
         """
-        storage = SagaExecutionStorage.from_config(config, **kwargs)
+        storage = DatabaseSagaExecutionRepository.from_config(config, **kwargs)
         return cls(storage=storage, **kwargs)
 
+    async def _setup(self) -> None:
+        await super()._setup()
+        await self.storage.setup()
+
+    async def _destroy(self) -> None:
+        await self.storage.destroy()
+        await super()._destroy()
+
     async def run(
         self,
         definition: Optional[Saga] = None,
@@ -161,7 +170,7 @@ async def _run_new(
         return await self._run(execution, **kwargs)
 
     async def _load_and_run(self, response: SagaResponse, **kwargs) -> Union[UUID, SagaExecution]:
-        execution = self.storage.load(response.uuid)
+        execution = await self.storage.load(response.uuid)
         return await self._run(execution, response=response, **kwargs)
 
     async def _run(
@@ -178,7 +187,7 @@ async def _run(
             else:
                 await self._run_with_pause_on_memory(execution, **kwargs)
         except SagaFailedExecutionException as exc:
-            self.storage.store(execution)
+            await self.storage.store(execution)
             if raise_on_error:
                 raise exc
             logger.exception(f"The execution identified by {execution.uuid!s} failed")
@@ -194,7 +203,7 @@ async def _run(
                 headers["related_services"] = ",".join(related_services)
 
         if execution.status == SagaStatus.Finished:
-            self.storage.delete(execution)
+            await self.storage.delete(execution)
 
         if return_execution:
             return execution
@@ -207,7 +216,7 @@ async def _run_with_pause_on_disk(self, execution: SagaExecution, autocommit: bo
             if autocommit:
                 await execution.commit(**kwargs)
         except SagaPausedExecutionStepException:
-            self.storage.store(execution)
+            await self.storage.store(execution)
         except SagaFailedExecutionException as exc:
             if autocommit:
                 await execution.reject(**kwargs)
@@ -225,7 +234,7 @@ async def _run_with_pause_on_memory(
                         await execution.execute(response=response, autocommit=False, **kwargs)
                     except SagaPausedExecutionStepException:
                         response = await self._get_response(broker, execution, **kwargs)
-                    self.storage.store(execution)
+                    await self.storage.store(execution)
             if autocommit:
                 await execution.commit(**kwargs)
         except SagaFailedExecutionException as exc: