Merge pull request #468 from minos-framework/issue-51-implement-aiopg-saga-execution-repository

#51 - Implement `aiopg`'s `SagaExecutionRepository`

Author: Sergio García Prado

.github/workflows/minos-database-aiopg-tests.yml

@@ -11,6 +11,7 @@ on:
       - 'packages/core/minos-microservice-aggregate/**'
       - 'packages/core/minos-microservice-networks/**'
       - 'packages/core/minos-microservice-transactions/**'
+      - 'packages/core/minos-microservice-saga/**'
       - 'packages/core/minos-microservice-common/**'
 
 jobs:

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

@@ -2,6 +2,10 @@
     ABC,
     abstractmethod,
 )
+from typing import (
+    Any,
+    Optional,
+)
 from uuid import (
     UUID,
 )
@@ -16,10 +20,35 @@ class SagaExecutionDatabaseOperationFactory(DatabaseOperationFactory, ABC):
     """Saga Execution Database Operation Factory class."""
 
     @abstractmethod
-    def build_store(self, uuid: UUID, **kwargs) -> DatabaseOperation:
+    def build_create(self) -> DatabaseOperation:
+        """Build the database operation to create the delta table.
+
+        :return: A ``DatabaseOperation`` instance.s
+        """
+
+    @abstractmethod
+    def build_store(
+        self,
+        uuid: UUID,
+        definition: dict[str, Any],
+        status: str,
+        executed_steps: list[dict[str, Any]],
+        paused_step: Optional[dict[str, Any]],
+        context: str,
+        already_rollback: bool,
+        user: Optional[UUID],
+        **kwargs
+    ) -> DatabaseOperation:
         """Build the database operation to store a saga execution.
 
         :param uuid: The identifier of the saga execution.
+        :param definition: The ``Saga`` definition.
+        :param context: The execution context.
+        :param status: The status of the execution.
+        :param executed_steps: The executed steps of the execution.
+        :param paused_step: The paused step of the execution.
+        :param already_rollback: ``True`` if already rollback of ``False`` otherwise.
+        :param user: The user that launched the execution.
         :param kwargs: The attributes of the saga execution.
         :return: A ``DatabaseOperation`` instance.
         """

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

@@ -36,6 +36,11 @@ def __init__(self, *args, database_key: Optional[tuple[str]] = None, **kwargs):
             database_key = ("saga",)
         super().__init__(*args, database_key=database_key, **kwargs)
 
+    async def _setup(self) -> None:
+        await super()._setup()
+        operation = self.database_operation_factory.build_create()
+        await self.execute_on_database(operation)
+
     async def _store(self, execution: SagaExecution) -> None:
         operation = self.database_operation_factory.build_store(**execution.raw)
         await self.execute_on_database(operation)

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

@@ -69,6 +69,19 @@ def __init__(
         *args,
         **kwargs,
     ):
+        """Initialize a Saga Execution.
+
+        :param definition: The ``Saga`` definition.
+        :param uuid: The identifier of the execution.
+        :param context: The execution context.
+        :param status: The status of the execution.
+        :param steps: The executed steps of the execution.
+        :param paused_step: The paused step of the execution.
+        :param already_rollback: ``True`` if already rollback of ``False`` otherwise.
+        :param user: The user that launched the execution.
+        :param args: Additional positional arguments.
+        :param kwargs: Additional named arguments.
+        """
         definition.validate()  # If not valid, raises an exception.
 
         if steps is None:
@@ -94,7 +107,7 @@ def from_raw(cls, raw: Union[dict[str, Any], SagaExecution], **kwargs) -> SagaEx
         if isinstance(raw, cls):
             return raw
 
-        raw = raw.copy()
+        raw = dict(raw)
 
         current = raw | kwargs
         current["definition"] = Saga.from_raw(current["definition"])
@@ -313,8 +326,8 @@ def raw(self) -> dict[str, Any]:
         :return: A ``dict`` instance.
         """
         return {
-            "definition": self.definition.raw,
             "uuid": str(self.uuid),
+            "definition": self.definition.raw,
             "status": self.status.raw,
             "executed_steps": [step.raw for step in self.executed_steps],
             "paused_step": None if self.paused_step is None else self.paused_step.raw,

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

@@ -35,6 +35,10 @@
 class MockedSagaExecutionDatabaseOperationFactory(SagaExecutionDatabaseOperationFactory):
     """For testing purposes"""
 
+    def build_create(self) -> DatabaseOperation:
+        """For testing purposes"""
+        return MockedDatabaseOperation("create_table")
+
     def build_store(self, uuid: UUID, **kwargs) -> DatabaseOperation:
         """For testing purposes"""
         return MockedDatabaseOperation("create_table")

packages/core/minos-microservice-saga/tests/test_saga/test_executions/test_repositories/test_database/test_factories.py

@@ -16,7 +16,8 @@ def test_abstract(self):
         self.assertTrue(issubclass(SagaExecutionDatabaseOperationFactory, (DatabaseOperationFactory, ABC)))
         # noinspection PyUnresolvedReferences
         self.assertEqual(
-            {"build_store", "build_load", "build_delete"}, SagaExecutionDatabaseOperationFactory.__abstractmethods__
+            {"build_create", "build_store", "build_load", "build_delete"},
+            SagaExecutionDatabaseOperationFactory.__abstractmethods__,
         )
 
 

packages/core/minos-microservice-saga/tests/test_saga/test_executions/test_saga/test_raw.py

@@ -1,4 +1,5 @@
 import unittest
+import uuid
 from unittest.mock import (
     MagicMock,
     patch,
@@ -29,19 +30,19 @@ def setUp(self) -> None:
         self.broker_publisher.send = self.publish_mock
 
     def test_from_raw(self):
-        with patch("uuid.uuid4", return_value=UUID("a74d9d6d-290a-492e-afcc-70607958f65d")):
+        with patch.object(uuid, "uuid4", return_value=UUID("a74d9d6d-290a-492e-afcc-70607958f65d")):
             expected = SagaExecution.from_definition(ADD_ORDER, user=self.user)
         observed = SagaExecution.from_raw(expected)
         self.assertEqual(expected, observed)
 
     def test_from_raw_without_user(self):
-        with patch("uuid.uuid4", return_value=UUID("a74d9d6d-290a-492e-afcc-70607958f65d")):
+        with patch.object(uuid, "uuid4", return_value=UUID("a74d9d6d-290a-492e-afcc-70607958f65d")):
             expected = SagaExecution.from_definition(ADD_ORDER)
         observed = SagaExecution.from_raw(expected)
         self.assertEqual(expected, observed)
 
     def test_created(self):
-        with patch("uuid.uuid4", return_value=UUID("a74d9d6d-290a-492e-afcc-70607958f65d")):
+        with patch.object(uuid, "uuid4", return_value=UUID("a74d9d6d-290a-492e-afcc-70607958f65d")):
             execution = SagaExecution.from_definition(ADD_ORDER, user=self.user)
 
         expected = {
@@ -87,7 +88,7 @@ def test_created(self):
         self.assertEqual(expected, observed)
 
     def test_created_without_user(self):
-        with patch("uuid.uuid4", return_value=UUID("a74d9d6d-290a-492e-afcc-70607958f65d")):
+        with patch.object(uuid, "uuid4", return_value=UUID("a74d9d6d-290a-492e-afcc-70607958f65d")):
             execution = SagaExecution.from_definition(ADD_ORDER)
 
         expected = {
@@ -182,7 +183,7 @@ async def test_partial_step(self):
             "uuid": "a74d9d6d-290a-492e-afcc-70607958f65d",
         }
 
-        with patch("uuid.uuid4", return_value=UUID("a74d9d6d-290a-492e-afcc-70607958f65d")):
+        with patch.object(uuid, "uuid4", return_value=UUID("a74d9d6d-290a-492e-afcc-70607958f65d")):
             expected = SagaExecution.from_definition(ADD_ORDER, user=self.user)
             with self.assertRaises(SagaPausedExecutionStepException):
                 await expected.execute()
@@ -268,7 +269,7 @@ async def test_executed_step(self):
             "uuid": "a74d9d6d-290a-492e-afcc-70607958f65d",
         }
 
-        with patch("uuid.uuid4", return_value=UUID("a74d9d6d-290a-492e-afcc-70607958f65d")):
+        with patch.object(uuid, "uuid4", return_value=UUID("a74d9d6d-290a-492e-afcc-70607958f65d")):
             expected = SagaExecution.from_definition(ADD_ORDER, user=self.user)
             with self.assertRaises(SagaPausedExecutionStepException):
                 await expected.execute()

packages/plugins/minos-database-aiopg/minos/plugins/aiopg/__init__.py

@@ -16,6 +16,7 @@
     AiopgDeltaDatabaseOperationFactory,
     AiopgLockDatabaseOperationFactory,
     AiopgManagementDatabaseOperationFactory,
+    AiopgSagaExecutionDatabaseOperationFactory,
     AiopgSnapshotDatabaseOperationFactory,
     AiopgSnapshotQueryDatabaseOperationBuilder,
     AiopgTransactionDatabaseOperationFactory,

packages/plugins/minos-database-aiopg/minos/plugins/aiopg/clients.py

@@ -27,6 +27,9 @@
     OperationalError,
     ProgrammingError,
 )
+from psycopg2.extras import (
+    DictCursor,
+)
 
 from minos.common import (
     CircuitBreakerMixin,
@@ -180,7 +183,7 @@ async def _execute_cursor(self, operation: str, parameters: dict):
         if not await self.is_connected():
             await self.recreate()
 
-        self._cursor = await self._connection.cursor(timeout=self._cursor_timeout)
+        self._cursor = await self._connection.cursor(timeout=self._cursor_timeout, cursor_factory=DictCursor)
         try:
             await self._cursor.execute(operation=operation, parameters=parameters)
         except OperationalError as exc:

packages/plugins/minos-database-aiopg/minos/plugins/aiopg/factories/__init__.py

@@ -14,6 +14,9 @@
     AiopgBrokerSubscriberDuplicateValidatorDatabaseOperationFactory,
     AiopgBrokerSubscriberQueueDatabaseOperationFactory,
 )
+from .saga import (
+    AiopgSagaExecutionDatabaseOperationFactory,
+)
 from .transactions import (
     AiopgTransactionDatabaseOperationFactory,
 )