Merge pull request #165 from minos-framework/issue-156-add-root-entity-get-all

#156 - Add `get_all` method to `SnapshotRepository`

Author: Sergio García Prado

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/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")

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

@@ -21,20 +21,20 @@
     UUID,
 )
 
-from minos.aggregate.actions import (
-    Action,
-)
 from minos.common import (
     DeclarativeModel,
 )
 
+from ..actions import (
+    Action,
+)
 from .fields import (
     FieldDiff,
     FieldDiffContainer,
 )
 
 if TYPE_CHECKING:
-    from minos.aggregate.entities import (
+    from ..entities import (
         RootEntity,
     )
 

packages/core/minos-microservice-aggregate/minos/aggregate/snapshots/abc.py

@@ -21,6 +21,7 @@
 )
 
 from ..queries import (
+    _TRUE_CONDITION,
     _Condition,
     _Ordering,
 )
@@ -64,6 +65,40 @@ async def get(self, name: str, uuid: UUID, transaction: Optional[TransactionEntr
     async def _get(self, *args, **kwargs) -> RootEntity:
         raise NotImplementedError
 
+    def get_all(
+        self,
+        name: str,
+        ordering: Optional[_Ordering] = None,
+        limit: Optional[int] = None,
+        streaming_mode: bool = False,
+        transaction: Optional[TransactionEntry] = None,
+        **kwargs,
+    ) -> AsyncIterator[RootEntity]:
+        """Get all ``RootEntity`` instances.
+
+        :param name: Class name of the ``RootEntity``.
+        :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 streaming_mode: If ``True`` return the values in streaming directly from the database (keep an open
+            database connection), otherwise preloads the full set of values on memory and then retrieves them.
+        :param transaction: The transaction within the operation is performed. If not any value is provided, then the
+            transaction is extracted from the context var. If not any transaction is being scoped then the query is
+            performed to the global snapshot.
+        :param kwargs: Additional named arguments.
+        :return: An asynchronous iterator that containing the ``RootEntity`` instances.
+        """
+        return self.find(
+            name,
+            _TRUE_CONDITION,
+            ordering=ordering,
+            limit=limit,
+            streaming_mode=streaming_mode,
+            transaction=transaction,
+            **kwargs,
+        )
+
     async def find(
         self,
         name: str,

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

@@ -7,17 +7,18 @@
     TypeVar,
 )
 
-from minos.aggregate.collections import (
-    IncrementalSet,
-)
-from minos.aggregate.exceptions import (
-    ValueObjectException,
-)
 from minos.common import (
     DeclarativeModel,
     Model,
 )
 
+from .collections import (
+    IncrementalSet,
+)
+from .exceptions import (
+    ValueObjectException,
+)
+
 
 class ValueObject(DeclarativeModel):
     """Value Object class."""