Merge pull request #178 from minos-framework/0.5.x

0.5.x

Author: Sergio García Prado

README.md

@@ -19,6 +19,7 @@ Minos is a framework which helps you create [reactive](https://www.reactivemanif
 ### Roadmap
 
 #### 0.6.x
+
 * [#78](https://github.com/minos-framework/minos-python/issues/78) Implement a circuit breaker for `minos-broker-kafka`.
 * [#87](https://github.com/minos-framework/minos-python/issues/87) Implement idempotency for `BrokerSubscriber` message processing.
 * [#100](https://github.com/minos-framework/minos-python/issues/100) Create the `minos-serializers-avro` plugin.
@@ -84,9 +85,54 @@ The required environment to run this quickstart is the following:
 * A `postgres` instance available at `localhost:5432` with the `foo_db` and `foobar_db` databases accessible with the `user:pass` credentials.
 * Two TCP sockets available to use at `localhost:4545` and `localhost:4546`.
 
+
+<details>
+  <summary>Click to show a <code>docker-compose.yml</code> that provides the <code>kafka</code> and <code>postgres</code> instances ready to be used!</summary>
+
+```yaml
+# docker-compose.yml
+version: "3.9"
+services:
+  zookeeper:
+    restart: always
+    image: wurstmeister/zookeeper:latest
+  kafka:
+    restart: always
+    image: wurstmeister/kafka:latest
+    ports:
+      - "9092:9092"
+    depends_on:
+      - zookeeper
+    environment:
+      KAFKA_ADVERTISED_HOST_NAME: kafka
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+  postgres:
+    restart: always
+    image: postgres:latest
+    ports:
+      - "5432:5432"
+    environment:
+      - POSTGRES_USER=user
+      - POSTGRES_PASSWORD=pass
+```
+
+Then, start the environment:
+
+```shell
+docker-compose up
+```
+
+To create the databases, just run the following command:
+
+```shell
+docker-compose exec postgres psql -U user -tc 'CREATE database foo_db'
+docker-compose exec postgres psql -U user -tc 'CREATE database foobar_db'
+```
+</details>
+
 Note that these parameters can be customized on the configuration files.
 
-###  Install the dependencies
+### Install the dependencies
 
 If you want to directly use `minos` without the command-line utility, the following command will install the needed packages:
 
@@ -100,7 +146,6 @@ pip install \
   minos-broker-kafka
 ```
 
-
 ### Configure a Microservice
 
 To keep things simpler, this quickstart will create a microservice assuming all the source code is stored on a single `foo/main.py` file. In addition to the source file, a `foo/config.yml` will contain all the configuration stuff.

packages/core/minos-microservice-aggregate/HISTORY.md

@@ -68,4 +68,12 @@
 
 ## 0.5.1 (2022-02-03)
 
-* Fix bug related with dependency specification.
+* Fix bug related with dependency specification.
+
+## 0.5.2 (2022-02-08)
+
+* Add `Condition.LIKE` operator to be used with the `find` method from `SnapshotRepository`.
+* Add `get_all` method to `RootEntity` and `SnapshotRepository` to get all the stored instance on the repository. 
+* Rename `SnapshotService` command topics to avoid collisions with application-level topics. 
+* Rename `TransactionService` command topics to avoid collisions with application-level topics. 
+* Minor changes.

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

@@ -1,6 +1,6 @@
 __author__ = "Minos Framework Devs"
 __email__ = "[email protected]"
-__version__ = "0.5.1"
+__version__ = "0.5.2"
 
 from .actions import (
     Action,

packages/core/minos-microservice-aggregate/minos/aggregate/entities/collections.py

@@ -18,10 +18,6 @@
     UUID,
 )
 
-from minos.aggregate.collections import (
-    IncrementalSet,
-    IncrementalSetDiff,
-)
 from minos.common import (
     DataDecoder,
     DataEncoder,
@@ -31,6 +27,11 @@
     SchemaEncoder,
 )
 
+from ..collections import (
+    IncrementalSet,
+    IncrementalSetDiff,
+)
+
 T = TypeVar("T", bound=Model)
 
 

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

@@ -22,28 +22,29 @@
     inject,
 )
 
-from minos.aggregate.events import (
+from minos.common import (
+    NULL_DATETIME,
+    NULL_UUID,
+    DeclarativeModel,
+    NotProvidedException,
+)
+
+from ..events import (
     Event,
     EventEntry,
     EventRepository,
     IncrementalFieldDiff,
 )
-from minos.aggregate.exceptions import (
+from ..exceptions import (
     EventRepositoryException,
 )
-from minos.aggregate.queries import (
+from ..queries import (
     _Condition,
     _Ordering,
 )
-from minos.aggregate.snapshots import (
+from ..snapshots import (
     SnapshotRepository,
 )
-from minos.common import (
-    NULL_DATETIME,
-    NULL_UUID,
-    DeclarativeModel,
-    NotProvidedException,
-)
 
 logger = logging.getLogger(__name__)
 
@@ -83,46 +84,74 @@ def __init__(
         version: int = 0,
         created_at: datetime = NULL_DATETIME,
         updated_at: datetime = NULL_DATETIME,
-        _repository: EventRepository = Provide["event_repository"],
-        _snapshot: SnapshotRepository = Provide["snapshot_repository"],
+        _event_repository: EventRepository = Provide["event_repository"],
+        _snapshot_repository: SnapshotRepository = Provide["snapshot_repository"],
         **kwargs,
     ):
 
         super().__init__(version, created_at, updated_at, *args, uuid=uuid, **kwargs)
 
-        if _repository is None or isinstance(_repository, Provide):
-            raise NotProvidedException("An event repository instance is required.")
-        if _snapshot is None or isinstance(_snapshot, Provide):
-            raise NotProvidedException("A snapshot instance is required.")
+        if _event_repository is None or isinstance(_event_repository, Provide):
+            raise NotProvidedException(f"A {EventRepository!r} instance is required.")
+        if _snapshot_repository is None or isinstance(_snapshot_repository, Provide):
+            raise NotProvidedException(f"A {SnapshotRepository!r} instance is required.")
 
-        self._repository = _repository
-        self._snapshot = _snapshot
+        self._event_repository = _event_repository
+        self._snapshot_repository = _snapshot_repository
 
     @classmethod
     @inject
     async def get(
-        cls: Type[T], uuid: UUID, _snapshot: SnapshotRepository = Provide["snapshot_repository"], **kwargs
+        cls: Type[T], uuid: UUID, *, _snapshot_repository: SnapshotRepository = Provide["snapshot_repository"], **kwargs
     ) -> T:
         """Get one instance from the database based on its identifier.
 
         :param uuid: The identifier of the instance.
-        :param _snapshot: Snapshot to be set to the root entity.
+        :param _snapshot_repository: Snapshot to be set to the root entity.
         :return: A ``RootEntity`` instance.
         """
-        if _snapshot is None or isinstance(_snapshot, Provide):
-            raise NotProvidedException("A snapshot instance is required.")
+        if _snapshot_repository is None or isinstance(_snapshot_repository, Provide):
+            raise NotProvidedException(f"A {SnapshotRepository!r} instance is required.")
 
         # noinspection PyTypeChecker
-        return await _snapshot.get(cls.classname, uuid, _snapshot=_snapshot, **kwargs)
+        return await _snapshot_repository.get(cls.classname, uuid, _snapshot_repository=_snapshot_repository, **kwargs)
 
     @classmethod
     @inject
-    async def find(
+    def get_all(
+        cls: Type[T],
+        ordering: Optional[_Ordering] = None,
+        limit: Optional[int] = None,
+        *,
+        _snapshot_repository: SnapshotRepository = Provide["snapshot_repository"],
+        **kwargs,
+    ) -> AsyncIterator[T]:
+        """Get all instance from the database.
+
+        :param ordering: Optional argument to return the instance with specific ordering strategy. The default behaviour
+            is to retrieve them without any order pattern.
+        :param limit: Optional argument to return only a subset of instances. The default behaviour is to return all the
+            instances that meet the given condition.
+        :param _snapshot_repository: Snapshot to be set to the root entity.
+        :return: A ``RootEntity`` instance.
+        """
+        if _snapshot_repository is None or isinstance(_snapshot_repository, Provide):
+            raise NotProvidedException(f"A {SnapshotRepository!r} instance is required.")
+
+        # noinspection PyTypeChecker
+        return _snapshot_repository.get_all(
+            cls.classname, ordering, limit, _snapshot_repository=_snapshot_repository, **kwargs
+        )
+
+    @classmethod
+    @inject
+    def find(
         cls: Type[T],
         condition: _Condition,
         ordering: Optional[_Ordering] = None,
         limit: Optional[int] = None,
-        _snapshot: SnapshotRepository = Provide["snapshot_repository"],
+        *,
+        _snapshot_repository: SnapshotRepository = Provide["snapshot_repository"],
         **kwargs,
     ) -> AsyncIterator[T]:
         """Find a collection of instances based on a given ``Condition``.
@@ -132,16 +161,15 @@ async def find(
             is to retrieve them without any order pattern.
         :param limit: Optional argument to return only a subset of instances. The default behaviour is to return all the
             instances that meet the given condition.
-        :param _snapshot: Snapshot to be set to the instances.
+        :param _snapshot_repository: Snapshot to be set to the instances.
         :return: An asynchronous iterator of ``RootEntity`` instances.
         """
-        if _snapshot is None or isinstance(_snapshot, Provide):
-            raise NotProvidedException("A snapshot instance is required.")
+        if _snapshot_repository is None or isinstance(_snapshot_repository, Provide):
+            raise NotProvidedException(f"A {SnapshotRepository!r} instance is required.")
         # noinspection PyTypeChecker
-        iterable = _snapshot.find(cls.classname, condition, ordering, limit, _snapshot=_snapshot, **kwargs)
-        # noinspection PyTypeChecker
-        async for instance in iterable:
-            yield instance
+        return _snapshot_repository.find(
+            cls.classname, condition, ordering, limit, _snapshot_repository=_snapshot_repository, **kwargs
+        )
 
     @classmethod
     async def create(cls: Type[T], *args, **kwargs) -> T:
@@ -171,7 +199,7 @@ async def create(cls: Type[T], *args, **kwargs) -> T:
         instance: T = cls(*args, **kwargs)
 
         event = Event.from_root_entity(instance)
-        entry = await instance._repository.submit(event)
+        entry = await instance._event_repository.submit(event)
 
         instance._update_from_repository_entry(entry)
 
@@ -201,12 +229,14 @@ async def update(self: T, **kwargs) -> T:
         for key, value in kwargs.items():
             setattr(self, key, value)
 
-        previous = await self.get(self.uuid, _repository=self._repository, _snapshot=self._snapshot)
+        previous = await self.get(
+            self.uuid, _event_repository=self._event_repository, _snapshot_repository=self._snapshot_repository
+        )
         event = self.diff(previous)
         if not len(event.fields_diff):
             return self
 
-        entry = await self._repository.submit(event)
+        entry = await self._event_repository.submit(event)
 
         self._update_from_repository_entry(entry)
 
@@ -234,17 +264,23 @@ async def save(self) -> None:
             if k not in {"uuid", "version", "created_at", "updated_at"}
         }
         if is_creation:
-            new = await self.create(**values, _repository=self._repository, _snapshot=self._snapshot)
+            new = await self.create(
+                **values, _event_repository=self._event_repository, _snapshot_repository=self._snapshot_repository
+            )
             self._fields |= new.fields
         else:
-            await self.update(**values, _repository=self._repository, _snapshot=self._snapshot)
+            await self.update(
+                **values, _event_repository=self._event_repository, _snapshot_repository=self._snapshot_repository
+            )
 
     async def refresh(self) -> None:
         """Refresh the state of the given instance.
 
         :return: This method does not return anything.
         """
-        new = await type(self).get(self.uuid, _repository=self._repository, _snapshot=self._snapshot)
+        new = await self.get(
+            self.uuid, _event_repository=self._event_repository, _snapshot_repository=self._snapshot_repository
+        )
         self._fields |= new.fields
 
     async def delete(self) -> None:
@@ -253,7 +289,7 @@ async def delete(self) -> None:
         :return: This method does not return anything.
         """
         event = Event.from_deleted_root_entity(self)
-        entry = await self._repository.submit(event)
+        entry = await self._event_repository.submit(event)
 
         self._update_from_repository_entry(entry)
 

packages/core/minos-microservice-aggregate/minos/aggregate/entities/refs/models.py

@@ -20,9 +20,6 @@
     inject,
 )
 
-from minos.aggregate.contextvars import (
-    IS_REPOSITORY_SERIALIZATION_CONTEXT_VAR,
-)
 from minos.common import (
     DataDecoder,
     DataEncoder,
@@ -39,6 +36,10 @@
     BrokerMessageV1Payload,
 )
 
+from ...contextvars import (
+    IS_REPOSITORY_SERIALIZATION_CONTEXT_VAR,
+)
+
 MT = TypeVar("MT", bound=Model)
 
 

packages/core/minos-microservice-aggregate/minos/aggregate/entities/refs/resolvers.py

@@ -65,8 +65,10 @@ async def resolve(self, data: Any, **kwargs) -> Any:
     async def _query(self, references: dict[str, set[UUID]]) -> dict[UUID, Model]:
         async with self.broker_pool.acquire() as broker:
             futures = (
-                broker.send(BrokerMessageV1(f"Get{name}s", BrokerMessageV1Payload({"uuids": uuids})))
-                for name, uuids in references.items()
+                broker.send(
+                    BrokerMessageV1(f"_Get{simplified_name}Snapshots", BrokerMessageV1Payload({"uuids": uuids}))
+                )
+                for simplified_name, uuids in references.items()
             )
             await gather(*futures)
 

packages/core/minos-microservice-aggregate/minos/aggregate/events/fields.py

@@ -22,19 +22,20 @@
     uuid4,
 )
 
-from minos.aggregate.actions import (
-    Action,
-)
-from minos.aggregate.collections import (
-    IncrementalSet,
-)
 from minos.common import (
     BucketModel,
     Field,
     Model,
     ModelType,
 )
 
+from ..actions import (
+    Action,
+)
+from ..collections import (
+    IncrementalSet,
+)
+
 logger = logging.getLogger(__name__)
 
 T = TypeVar("T")