ISSUE #98

* Add support for multiple databases.

Author: Sergio García Prado

packages/core/minos-microservice-aggregate/minos/aggregate/events/repositories/database/impl.py

@@ -9,7 +9,6 @@
 
 from minos.common import (
     NULL_UUID,
-    Config,
     DatabaseMixin,
     DatabaseOperation,
     IntegrityException,
@@ -32,9 +31,10 @@
 class DatabaseEventRepository(DatabaseMixin[EventDatabaseOperationFactory], EventRepository):
     """Database-based implementation of the event repository class."""
 
-    @classmethod
-    def _from_config(cls, config: Config, **kwargs) -> Optional[EventRepository]:
-        return super()._from_config(config, database_key=None, **kwargs)
+    def __init__(self, *args, database_key: Optional[tuple[str]] = None, **kwargs):
+        if database_key is None:
+            database_key = ("aggregate", "event")
+        super().__init__(*args, database_key=database_key, **kwargs)
 
     async def _setup(self):
         """Setup miscellaneous repository things.

packages/core/minos-microservice-aggregate/minos/aggregate/snapshots/repositories/database/impl.py

@@ -15,7 +15,6 @@
 
 from minos.common import (
     NULL_UUID,
-    Config,
     DatabaseMixin,
     Inject,
     NotProvidedException,
@@ -71,9 +70,12 @@ def __init__(
         *args,
         event_repository: EventRepository,
         transaction_repository: TransactionRepository,
+        database_key: Optional[tuple[str]] = None,
         **kwargs,
     ):
-        super().__init__(*args, **kwargs)
+        if database_key is None:
+            database_key = ("aggregate", "snapshot")
+        super().__init__(*args, database_key=database_key, **kwargs)
 
         if event_repository is None:
             raise NotProvidedException("An event repository instance is required.")
@@ -84,10 +86,6 @@ def __init__(
         self._event_repository = event_repository
         self._transaction_repository = transaction_repository
 
-    @classmethod
-    def _from_config(cls, config: Config, **kwargs) -> DatabaseSnapshotRepository:
-        return cls(database_key=None, **kwargs)
-
     async def _setup(self) -> None:
         operation = self.operation_factory.build_create()
         await self.submit_query(operation)

packages/core/minos-microservice-aggregate/minos/aggregate/transactions/repositories/database/impl.py

@@ -8,7 +8,6 @@
 )
 
 from minos.common import (
-    Config,
     DatabaseMixin,
     ProgrammingException,
 )
@@ -30,9 +29,10 @@
 class DatabaseTransactionRepository(DatabaseMixin[TransactionDatabaseOperationFactory], TransactionRepository):
     """Database Transaction Repository class."""
 
-    @classmethod
-    def _from_config(cls, config: Config, **kwargs) -> DatabaseTransactionRepository:
-        return super()._from_config(config, database_key=None, **kwargs)
+    def __init__(self, *args, database_key: Optional[tuple[str]] = None, **kwargs):
+        if database_key is None:
+            database_key = ("aggregate", "transaction")
+        super().__init__(*args, database_key=database_key, **kwargs)
 
     async def _setup(self):
         operation = self.operation_factory.build_create()

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

@@ -113,6 +113,7 @@
 from .pools import (
     MinosPool,
     Pool,
+    PoolException,
     PoolFactory,
 )
 from .ports import (

packages/core/minos-microservice-common/minos/common/database/mixins.py

@@ -1,3 +1,6 @@
+from contextlib import (
+    suppress,
+)
 from typing import (
     AsyncIterator,
     Generic,
@@ -14,6 +17,7 @@
     Inject,
 )
 from ..pools import (
+    PoolException,
     PoolFactory,
 )
 from ..setup import (
@@ -39,15 +43,16 @@ def __init__(
         self,
         database_pool: Optional[DatabaseClientPool] = None,
         pool_factory: Optional[PoolFactory] = None,
-        database_key: Optional[str] = None,
+        database_key: Optional[tuple[str]] = None,
         operation_factory: Optional[GenericDatabaseOperationFactory] = None,
         operation_factory_cls: Optional[type[GenericDatabaseOperationFactory]] = None,
         *args,
         **kwargs,
     ):
         super().__init__(*args, **kwargs, pool_factory=pool_factory)
+
         if database_pool is None and pool_factory is not None:
-            database_pool = pool_factory.get_pool(type_="database", identifier=database_key)
+            database_pool = self._get_pool_from_factory(pool_factory, database_key)
 
         if not isinstance(database_pool, DatabaseClientPool):
             raise NotProvidedException(f"A {DatabaseClientPool!r} instance is required. Obtained: {database_pool}")
@@ -62,6 +67,17 @@ def __init__(
 
         self._operation_factory = operation_factory
 
+    @staticmethod
+    def _get_pool_from_factory(pool_factory: PoolFactory, database_key: Optional[tuple[str]]):
+        if database_key is None:
+            database_key = tuple()
+
+        for identifier in database_key:
+            with suppress(PoolException):
+                return pool_factory.get_pool(type_="database", identifier=identifier)
+
+        return pool_factory.get_pool(type_="database")
+
     @property
     def operation_factory(self) -> Optional[GenericDatabaseOperationFactory]:
         """Get the operation factory if any.

packages/core/minos-microservice-common/minos/common/pools.py

@@ -29,6 +29,10 @@
 from .config import (
     Config,
 )
+from .exceptions import (
+    MinosConfigException,
+    MinosException,
+)
 from .injections import (
     Injectable,
 )
@@ -87,7 +91,10 @@ def get_pool(self, type_: str, identifier: Optional[str] = None, **kwargs) -> Po
     def _create_pool(self, type_: str, **kwargs) -> Pool:
         # noinspection PyTypeChecker
         pool_cls = self._get_pool_cls(type_)
-        pool = pool_cls.from_config(self._config, **kwargs)
+        try:
+            pool = pool_cls.from_config(self._config, **kwargs)
+        except MinosConfigException:
+            raise PoolException("The pool could not be built.")
         return pool
 
     def _get_pool_cls(self, type_: str) -> type[Pool]:
@@ -97,7 +104,7 @@ def _get_pool_cls(self, type_: str) -> type[Pool]:
             pool_cls = self._config.get_pools().get("types", dict()).get(type_)
 
         if pool_cls is None:
-            raise ValueError(
+            raise PoolException(
                 f"There is not any provided {type!r} to build pools that matches the given type: {type_!r}"
             )
 
@@ -175,3 +182,7 @@ class MinosPool(Pool, Generic[P], ABC):
     def __init__(self, *args, **kwargs):
         warnings.warn(f"{MinosPool!r} has been deprecated. Use {Pool} instead.", DeprecationWarning)
         super().__init__(*args, **kwargs)
+
+
+class PoolException(MinosException):
+    """Exception to be raised when some problem related with a pool happens."""

packages/core/minos-microservice-common/tests/config/v2.yml

@@ -15,6 +15,7 @@ databases:
   saga:
     path: "./order.lmdb"
   query:
+    client: minos.common.testing.MockedDatabaseClient
     database: order_query_db
     user: minos
     password: min0s

packages/core/minos-microservice-common/tests/test_common/test_database/test_mixins.py

@@ -33,6 +33,13 @@ async def test_constructor_with_pool_factory(self):
         # noinspection PyUnresolvedReferences
         self.assertEqual(pool_factory.get_pool("database"), database.database_pool)
 
+    async def test_constructor_with_pool_factory_and_database_key(self):
+        pool_factory = PoolFactory(self.config, {"database": DatabaseClientPool})
+        # noinspection PyTypeChecker
+        database = DatabaseMixin(pool_factory=pool_factory, database_key=("query", "unknown"))
+        # noinspection PyUnresolvedReferences
+        self.assertEqual(pool_factory.get_pool("database", "query"), database.database_pool)
+
     async def test_constructor_raises(self):
         with self.assertRaises(NotProvidedException):
             # noinspection PyArgumentEqualDefault

packages/core/minos-microservice-common/tests/test_common/test_pools.py

@@ -10,15 +10,18 @@
 )
 from unittest.mock import (
     MagicMock,
+    patch,
 )
 
 from aiomisc import (
     PoolBase,
 )
 
 from minos.common import (
+    MinosConfigException,
     MinosPool,
     Pool,
+    PoolException,
     PoolFactory,
     SetupMixin,
 )
@@ -55,10 +58,15 @@ def test_get_pool_with_key(self):
         self.assertEqual(lock_a, self.factory.get_pool("lock", "a"))
         self.assertEqual(lock_b, self.factory.get_pool("lock", "b"))
 
-    def test_get_pool_raises(self):
-        with self.assertRaises(ValueError):
+    def test_get_pool_cls_raises(self):
+        with self.assertRaises(PoolException):
             self.factory.get_pool("something")
 
+    def test_get_pool_identifier_raises(self):
+        with patch.object(SetupMixin, "from_config", side_effect=MinosConfigException("")):
+            with self.assertRaises(PoolException):
+                self.factory.get_pool("database")
+
 
 class TestPool(unittest.IsolatedAsyncioTestCase):
     def test_abstract(self):

packages/core/minos-microservice-networks/minos/networks/brokers/collections/queues/database/impl.py

@@ -65,9 +65,12 @@ def __init__(
         *args,
         retry: Optional[int] = None,
         records: Optional[int] = None,
+        database_key: Optional[tuple[str]] = None,
         **kwargs,
     ):
-        super().__init__(*args, **kwargs)
+        if database_key is None:
+            database_key = ("broker",)
+        super().__init__(*args, database_key=database_key, **kwargs)
 
         if retry is None:
             retry = 2