#367 - Add DatabaseClient

packages/core/minos-microservice-aggregate/minos/aggregate/events/repositories/pg.py

@@ -10,9 +10,6 @@
     UUID,
 )
 
-from psycopg2 import (
-    IntegrityError,
-)
 from psycopg2.sql import (
     SQL,
     Composable,
@@ -24,6 +21,7 @@
     NULL_UUID,
     Config,
     DatabaseMixin,
+    IntegrityException,
 )
 
 from ...exceptions import (
@@ -42,7 +40,7 @@ class PostgreSqlEventRepository(DatabaseMixin, EventRepository):
 
     @classmethod
     def _from_config(cls, config: Config, **kwargs) -> Optional[EventRepository]:
-        return super()._from_config(config, **config.get_database_by_name("event"), **kwargs)
+        return super()._from_config(config, database_key=None, **kwargs)
 
     async def _setup(self):
         """Setup miscellaneous repository thing.
@@ -65,7 +63,7 @@ async def _submit(self, entry: EventEntry, **kwargs) -> EventEntry:
 
         try:
             response = await self.submit_query_and_fetchone(query, params, lock=lock)
-        except IntegrityError:
+        except IntegrityException:
             raise EventRepositoryConflictException(
                 f"{entry!r} could not be submitted due to a key (uuid, version, transaction) collision",
                 await self.offset,

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

@@ -18,7 +18,7 @@ class PostgreSqlSnapshotSetup(DatabaseMixin):
 
     @classmethod
     def _from_config(cls: Type[T], config: Config, **kwargs) -> T:
-        return cls(**config.get_database_by_name("snapshot"), **kwargs)
+        return cls(database_key=None, **kwargs)
 
     async def _setup(self) -> None:
         await self.submit_query('CREATE EXTENSION IF NOT EXISTS "uuid-ossp";', lock="uuid-ossp")

packages/core/minos-microservice-aggregate/minos/aggregate/snapshots/pg/api.py

@@ -51,7 +51,7 @@ def _from_config(cls, config: Config, **kwargs) -> PostgreSqlSnapshotRepository:
         if "writer" not in kwargs:
             kwargs["writer"] = PostgreSqlSnapshotWriter.from_config(config, **kwargs)
 
-        return cls(**config.get_database_by_name("snapshot"), **kwargs)
+        return cls(database_key=None, **kwargs)
 
     async def _setup(self) -> None:
         await self.writer.setup()

packages/core/minos-microservice-aggregate/minos/aggregate/snapshots/pg/readers.py

@@ -128,15 +128,10 @@ async def find_entries(
         qb = PostgreSqlSnapshotQueryBuilder(name, condition, ordering, limit, transaction_uuids, exclude_deleted)
         query, parameters = qb.build()
 
-        async with self.cursor() as cursor:
-            # noinspection PyTypeChecker
-            await cursor.execute(query, parameters)
-            if streaming_mode:
-                async for row in cursor:
-                    # noinspection PyArgumentList
-                    yield SnapshotEntry(*row)
-                return
-            else:
-                rows = await cursor.fetchall()
-        for row in rows:
-            yield SnapshotEntry(*row)
+        async_iterable = self.submit_query_and_iter(query, parameters)
+        if streaming_mode:
+            async for row in async_iterable:
+                yield SnapshotEntry(*row)
+        else:
+            for row in [row async for row in async_iterable]:
+                yield SnapshotEntry(*row)

packages/core/minos-microservice-aggregate/minos/aggregate/transactions/repositories/pg.py

@@ -34,7 +34,7 @@ class PostgreSqlTransactionRepository(DatabaseMixin, TransactionRepository):
 
     @classmethod
     def _from_config(cls, config: Config, **kwargs) -> Optional[PostgreSqlTransactionRepository]:
-        return super()._from_config(config, **config.get_database_by_name("transaction"), **kwargs)
+        return super()._from_config(config, database_key=None, **kwargs)
 
     async def _setup(self):
         await self.submit_query('CREATE EXTENSION IF NOT EXISTS "uuid-ossp";', lock="uuid-ossp")

packages/core/minos-microservice-aggregate/tests/test_aggregate/test_events/test_repositories/test_pg.py

@@ -1,12 +1,11 @@
 import unittest
 
-import aiopg
-
 from minos.aggregate import (
     EventRepository,
     PostgreSqlEventRepository,
 )
 from minos.common import (
+    AiopgDatabaseClient,
     DatabaseClientPool,
 )
 from minos.common.testing import (
@@ -18,6 +17,7 @@
 )
 
 
+# noinspection SqlNoDataSourceInspection
 class TestPostgreSqlEventRepositorySubmit(EventRepositorySubmitTestCase, PostgresAsyncTestCase):
     __test__ = True
 
@@ -30,20 +30,18 @@ def test_constructor(self):
         pool = DatabaseClientPool.from_config(self.config)
         repository = PostgreSqlEventRepository(pool)
         self.assertIsInstance(repository, PostgreSqlEventRepository)
-        self.assertEqual(pool, repository.pool)
+        self.assertIsInstance(repository.pool, DatabaseClientPool)
 
     def test_from_config(self):
         repository = PostgreSqlEventRepository.from_config(self.config)
-        repository_config = self.config.get_database_by_name("event")
-        self.assertEqual(repository_config["database"], repository.pool.database)
+        self.assertIsInstance(repository.pool, DatabaseClientPool)
 
     async def test_setup(self):
-        async with aiopg.connect(**self.config.get_default_database()) as connection:
-            async with connection.cursor() as cursor:
-                await cursor.execute(
-                    "SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'aggregate_event');"
-                )
-                response = (await cursor.fetchone())[0]
+        async with AiopgDatabaseClient(**self.config.get_default_database()) as client:
+            await client.execute(
+                "SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'aggregate_event');"
+            )
+            response = (await client.fetch_one())[0]
         self.assertTrue(response)
 
 

packages/core/minos-microservice-aggregate/tests/test_aggregate/test_snapshots/test_pg/test_abc.py

@@ -1,10 +1,11 @@
 import unittest
 
-import aiopg
-
 from minos.aggregate import (
     PostgreSqlSnapshotSetup,
 )
+from minos.common import (
+    AiopgDatabaseClient,
+)
 from minos.common.testing import (
     PostgresAsyncTestCase,
 )
@@ -13,27 +14,25 @@
 )
 
 
+# noinspection SqlNoDataSourceInspection
 class TestPostgreSqlSnapshotSetup(AggregateTestCase, PostgresAsyncTestCase):
     async def test_setup_snapshot_table(self):
         async with PostgreSqlSnapshotSetup.from_config(self.config):
-            async with aiopg.connect(**self.config.get_default_database()) as connection:
-                async with connection.cursor() as cursor:
-                    await cursor.execute(
-                        "SELECT EXISTS (SELECT FROM pg_tables "
-                        "WHERE schemaname = 'public' AND tablename = 'snapshot');"
-                    )
-                    observed = (await cursor.fetchone())[0]
+            async with AiopgDatabaseClient(**self.config.get_default_database()) as client:
+                await client.execute(
+                    "SELECT EXISTS (SELECT FROM pg_tables WHERE schemaname = 'public' AND tablename = 'snapshot');"
+                )
+                observed = (await client.fetch_one())[0]
         self.assertEqual(True, observed)
 
     async def test_setup_snapshot_aux_offset_table(self):
         async with PostgreSqlSnapshotSetup.from_config(self.config):
-            async with aiopg.connect(**self.config.get_default_database()) as connection:
-                async with connection.cursor() as cursor:
-                    await cursor.execute(
-                        "SELECT EXISTS (SELECT FROM pg_tables WHERE "
-                        "schemaname = 'public' AND tablename = 'snapshot_aux_offset');"
-                    )
-                    observed = (await cursor.fetchone())[0]
+            async with AiopgDatabaseClient(**self.config.get_default_database()) as client:
+                await client.execute(
+                    "SELECT EXISTS (SELECT FROM pg_tables WHERE "
+                    "schemaname = 'public' AND tablename = 'snapshot_aux_offset');"
+                )
+                observed = (await client.fetch_one())[0]
         self.assertEqual(True, observed)
 
 

packages/core/minos-microservice-aggregate/tests/test_aggregate/test_snapshots/test_pg/test_queries.py

@@ -7,7 +7,6 @@
     uuid4,
 )
 
-import aiopg
 from psycopg2.extras import (
     Json,
 )
@@ -29,6 +28,7 @@
 )
 from minos.common import (
     NULL_UUID,
+    AiopgDatabaseClient,
 )
 from minos.common.testing import (
     PostgresAsyncTestCase,
@@ -426,8 +426,8 @@ async def test_build_complex(self):
         self.assertEqual(self._flatten_parameters(expected_parameters), self._flatten_parameters(observed[1]))
 
     async def _flatten_query(self, query) -> str:
-        async with aiopg.connect(**self.config.get_default_database()) as connection:
-            return query.as_string(connection.raw)
+        async with AiopgDatabaseClient(**self.config.get_default_database()) as client:
+            return query.as_string(client.connection.raw)
 
     @staticmethod
     def _flatten_parameters(parameters) -> dict:

packages/core/minos-microservice-aggregate/tests/test_aggregate/test_snapshots/test_pg/test_readers.py

@@ -21,6 +21,9 @@
     TransactionEntry,
     TransactionStatus,
 )
+from minos.common import (
+    DatabaseClientPool,
+)
 from minos.common.testing import (
     PostgresAsyncTestCase,
 )
@@ -94,8 +97,7 @@ def test_type(self):
 
     def test_from_config(self):
         reader = PostgreSqlSnapshotReader.from_config(self.config)
-        snapshot_config = self.config.get_database_by_name("snapshot")
-        self.assertEqual(snapshot_config["database"], reader.pool.database)
+        self.assertIsInstance(reader.pool, DatabaseClientPool)
 
     async def test_find_by_uuid(self):
         condition = Condition.IN("uuid", [self.uuid_2, self.uuid_3])

packages/core/minos-microservice-aggregate/tests/test_aggregate/test_snapshots/test_pg/test_writers.py

@@ -26,6 +26,7 @@
     TransactionStatus,
 )
 from minos.common import (
+    DatabaseClientPool,
     NotProvidedException,
     current_datetime,
 )
@@ -101,8 +102,7 @@ def test_type(self):
         self.assertTrue(issubclass(PostgreSqlSnapshotWriter, PostgreSqlSnapshotSetup))
 
     def test_from_config(self):
-        snapshot_config = self.config.get_database_by_name("snapshot")
-        self.assertEqual(snapshot_config["database"], self.writer.pool.database)
+        self.assertIsInstance(self.writer.pool, DatabaseClientPool)
 
     def test_from_config_raises(self):
         with self.assertRaises(NotProvidedException):

packages/core/minos-microservice-aggregate/tests/test_aggregate/test_transactions/test_repositories/test_pg.py

@@ -3,8 +3,6 @@
     uuid4,
 )
 
-import aiopg
-
 from minos.aggregate import (
     PostgreSqlTransactionRepository,
     TransactionEntry,
@@ -13,6 +11,7 @@
     TransactionStatus,
 )
 from minos.common import (
+    AiopgDatabaseClient,
     DatabaseClientPool,
 )
 from minos.common.testing import (
@@ -23,6 +22,7 @@
 )
 
 
+# noinspection SqlNoDataSourceInspection
 class TestPostgreSqlTransactionRepository(AggregateTestCase, PostgresAsyncTestCase):
     def setUp(self) -> None:
         super().setUp()
@@ -50,16 +50,14 @@ def test_constructor(self):
 
     def test_from_config(self):
         repository = PostgreSqlTransactionRepository.from_config(self.config)
-        repository_config = self.config.get_database_by_name("event")
-        self.assertEqual(repository_config["database"], repository.pool.database)
+        self.assertIsInstance(repository.pool, DatabaseClientPool)
 
     async def test_setup(self):
-        async with aiopg.connect(**self.config.get_default_database()) as connection:
-            async with connection.cursor() as cursor:
-                await cursor.execute(
-                    "SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'aggregate_transaction');"
-                )
-                response = (await cursor.fetchone())[0]
+        async with AiopgDatabaseClient(**self.config.get_default_database()) as client:
+            await client.execute(
+                "SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'aggregate_transaction');"
+            )
+            response = (await client.fetch_one())[0]
         self.assertTrue(response)
 
     async def test_submit(self):

packages/core/minos-microservice-aggregate/tests/utils.py

@@ -101,8 +101,11 @@ def __init__(self, key=None, *args, **kwargs):
             key = "fake"
         super().__init__(key, *args, **kwargs)
 
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
-        return
+    async def acquire(self) -> None:
+        """For testing purposes."""
+
+    async def release(self):
+        """For testing purposes."""
 
 
 class FakeLockPool(LockPool):

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

@@ -14,14 +14,20 @@
     MinosConfig,
 )
 from .database import (
+    AiopgDatabaseClient,
+    DatabaseClient,
+    DatabaseClientBuilder,
+    DatabaseClientException,
     DatabaseClientPool,
     DatabaseLock,
     DatabaseLockPool,
     DatabaseMixin,
+    IntegrityException,
     PostgreSqlLock,
     PostgreSqlLockPool,
     PostgreSqlMinosDatabase,
     PostgreSqlPool,
+    UnableToConnectException,
 )
 from .datetime import (
     NULL_DATETIME,

packages/core/minos-microservice-common/minos/common/config/v1.py

@@ -282,6 +282,8 @@ def _get_database_by_name(self, prefix: str):
         data = self.get_by_key(prefix)
         data.pop("records", None)
         data.pop("retry", None)
+        if "client" in data:
+            data["client"] = import_module(data["client"])
         return data
 
     def _get_discovery(self) -> dict[str, Any]:

packages/core/minos-microservice-common/minos/common/config/v2.py

@@ -99,6 +99,9 @@ def _get_injections(self) -> list[Union[InjectableMixin, type[InjectableMixin]]]
 
     def _get_databases(self) -> dict[str, dict[str, Any]]:
         data = deepcopy(self.get_by_key("databases"))
+        for database in data.values():
+            if "client" in database:
+                database["client"] = import_module(database["client"])
         return data
 
     def _get_interfaces(self) -> dict[str, dict[str, Any]]:

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

@@ -2,6 +2,14 @@
     DatabaseMixin,
     PostgreSqlMinosDatabase,
 )
+from .clients import (
+    AiopgDatabaseClient,
+    DatabaseClient,
+    DatabaseClientBuilder,
+    DatabaseClientException,
+    IntegrityException,
+    UnableToConnectException,
+)
 from .locks import (
     DatabaseLock,
     PostgreSqlLock,