Merge pull request #463 from minos-framework/0.8.0

0.8.0.dev2

.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:

.github/workflows/minos-microservice-aggregate-tests.yml

@@ -10,6 +10,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:

docs/index.md

@@ -29,9 +29,12 @@ Minos is a framework which helps you create [reactive](https://www.reactivemanif
    minos.cqrs
    minos.networks
    minos.saga
+   minos.transactions
    minos.plugins.kafka
    minos.plugins.rabbitmq
    minos.plugins.minos_discovery
+   minos.plugins.aiopg
+   minos.plugins.lmdb
    minos.plugins.kong
    minos.plugins.aiohttp
    minos.plugins.graphql

packages/core/minos-microservice-aggregate/minos/aggregate/deltas/models.py

@@ -63,7 +63,9 @@ def simplified_name(self) -> str:
         return self.name.rsplit(".", 1)[-1]
 
     def __lt__(self, other: Any) -> bool:
-        return isinstance(other, type(self)) and self.version < other.version
+        return isinstance(other, type(self)) and (
+            (self.uuid == other.uuid and self.version < other.version) or (self.created_at < other.created_at)
+        )
 
     def __getitem__(self, item: str) -> Any:
         try:

packages/core/minos-microservice-aggregate/minos/aggregate/impl.py

@@ -24,6 +24,9 @@
     BrokerMessageV1Payload,
     BrokerPublisher,
 )
+from minos.saga import (
+    SagaManager,
+)
 from minos.transactions import (
     TransactionRepository,
 )
@@ -56,6 +59,7 @@ class Aggregate(Generic[RT], SetupMixin):
     snapshot_repository: SnapshotRepository
     repository: EntityRepository
     broker_publisher: BrokerPublisher
+    saga_manager: SagaManager
 
     @Inject()
     def __init__(
@@ -64,6 +68,7 @@ def __init__(
         delta_repository: DeltaRepository,
         snapshot_repository: SnapshotRepository,
         broker_publisher: BrokerPublisher,
+        saga_manager: SagaManager,
         *args,
         **kwargs,
     ):
@@ -75,6 +80,8 @@ def __init__(
             raise NotProvidedException(f"A {SnapshotRepository!r} object must be provided.")
         if broker_publisher is None:
             raise NotProvidedException(f"A {BrokerPublisher!r} object must be provided.")
+        if saga_manager is None:
+            raise NotProvidedException(f"A {SagaManager!r} object must be provided.")
 
         super().__init__(*args, **kwargs)
 
@@ -86,6 +93,7 @@ def __init__(
         self.delta_repository = delta_repository
         self.snapshot_repository = snapshot_repository
         self.broker_publisher = broker_publisher
+        self.saga_manager = saga_manager
 
     @classmethod
     def _from_config(cls, config: Config, **kwargs) -> Aggregate:

packages/core/minos-microservice-aggregate/minos/aggregate/queries.py

@@ -146,6 +146,11 @@ def _evaluate(self, value: Model) -> bool:
         return self._get_field(value) in self.parameter
 
 
+class _ContainsCondition(_SimpleCondition):
+    def _evaluate(self, value: Model) -> bool:
+        return self.parameter in self._get_field(value)
+
+
 class _LikeCondition(_SimpleCondition):
     def _evaluate(self, value: Model) -> bool:
         return bool(self._pattern.fullmatch(self._get_field(value)))
@@ -205,6 +210,7 @@ class Condition:
     EQUAL = _EqualCondition
     NOT_EQUAL = _NotEqualCondition
     IN = _InCondition
+    CONTAINS = _ContainsCondition
     LIKE = _LikeCondition
 
 

packages/core/minos-microservice-aggregate/minos/aggregate/testing/snapshots/repositories/testcases.py

@@ -52,6 +52,16 @@ class SnapshotRepositoryTestCase(MinosTestCase, ABC):
 
     snapshot_repository: SnapshotRepository
 
+    class NumbersList(Entity):
+        """For testing purposes"""
+
+        numbers: list[int]
+
+    class Number(Entity):
+        """For testing purposes"""
+
+        value: int
+
     class Owner(Entity):
         """For testing purposes"""
 
@@ -380,6 +390,52 @@ async def test_find_by_uuid(self):
         ]
         self.assertEqual(expected, observed)
 
+    async def test_find_contains(self):
+        a = FieldDiffContainer([FieldDiff("numbers", list[int], [1, 2, 3])])
+        b = FieldDiffContainer([FieldDiff("numbers", list[int], [4, 5, 6])])
+        c = FieldDiffContainer([FieldDiff("numbers", list[int], [3, 8, 9])])
+        await self.delta_repository.create(
+            DeltaEntry(name=self.NumbersList.classname, data=a.avro_bytes, uuid=self.uuid_1)
+        )
+        await self.delta_repository.create(
+            DeltaEntry(name=self.NumbersList.classname, data=b.avro_bytes, uuid=self.uuid_2)
+        )
+        await self.delta_repository.create(
+            DeltaEntry(name=self.NumbersList.classname, data=c.avro_bytes, uuid=self.uuid_3)
+        )
+        await self.synchronize()
+
+        condition = Condition.CONTAINS("numbers", 3)
+
+        iterable = self.snapshot_repository.find(self.NumbersList, condition, ordering=Ordering.ASC("updated_at"))
+        observed = [v async for v in iterable]
+
+        expected = [
+            await self.snapshot_repository.get(self.NumbersList, self.uuid_1),
+            await self.snapshot_repository.get(self.NumbersList, self.uuid_3),
+        ]
+        self.assertEqual(expected, observed)
+
+    async def test_find_equal(self):
+        a = FieldDiffContainer([FieldDiff("value", int, 1)])
+        b = FieldDiffContainer([FieldDiff("value", int, 2)])
+        c = FieldDiffContainer([FieldDiff("value", int, 1)])
+        await self.delta_repository.create(DeltaEntry(name=self.Number.classname, data=a.avro_bytes, uuid=self.uuid_1))
+        await self.delta_repository.create(DeltaEntry(name=self.Number.classname, data=b.avro_bytes, uuid=self.uuid_2))
+        await self.delta_repository.create(DeltaEntry(name=self.Number.classname, data=c.avro_bytes, uuid=self.uuid_3))
+        await self.synchronize()
+
+        condition = Condition.EQUAL("value", 1)
+
+        iterable = self.snapshot_repository.find(self.Number, condition, ordering=Ordering.ASC("updated_at"))
+        observed = [v async for v in iterable]
+
+        expected = [
+            await self.snapshot_repository.get(self.Number, self.uuid_1),
+            await self.snapshot_repository.get(self.Number, self.uuid_3),
+        ]
+        self.assertEqual(expected, observed)
+
     async def test_find_with_transaction(self):
         await self.populate_and_synchronize()
         condition = Condition.IN("uuid", [self.uuid_2, self.uuid_3])