ISSUE #98

* Rename `submit_query` as `execute_on_database`.
* Rename `submit_query_and_fetchone` as `execute_on_database_and_fetch_one`.
* Rename `submit_query_and_fetchall` as `execute_on_database_and_fetch_all`.

Author: Sergio García Prado

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

@@ -41,14 +41,14 @@ async def _setup(self):
 
         :return: This method does not return anything.
         """
-        operation = self.operation_factory.build_create()
-        await self.submit_query(operation)
+        operation = self.database_operation_factory.build_create()
+        await self.execute_on_database(operation)
 
     async def _submit(self, entry: EventEntry, **kwargs) -> EventEntry:
         operation = await self._build_submit_operation(entry)
 
         try:
-            response = await self.submit_query_and_fetchone(operation)
+            response = await self.execute_on_database_and_fetch_one(operation)
         except IntegrityException:
             raise EventRepositoryConflictException(
                 f"{entry!r} could not be submitted due to a key (uuid, version, transaction) collision",
@@ -69,15 +69,17 @@ async def _build_submit_operation(self, entry: EventEntry) -> DatabaseOperation:
         else:
             transaction_uuids = (NULL_UUID,)
 
-        return self.operation_factory.build_submit(transaction_uuids=transaction_uuids, **entry.as_raw(), lock=lock)
+        return self.database_operation_factory.build_submit(
+            transaction_uuids=transaction_uuids, **entry.as_raw(), lock=lock
+        )
 
     async def _select(self, streaming_mode: Optional[bool] = None, **kwargs) -> AsyncIterator[EventEntry]:
-        operation = self.operation_factory.build_query(**kwargs)
-        async for row in self.submit_query_and_iter(operation, streaming_mode=streaming_mode):
+        operation = self.database_operation_factory.build_query(**kwargs)
+        async for row in self.execute_on_database_and_fetch_all(operation, streaming_mode=streaming_mode):
             yield EventEntry(*row)
 
     @property
     async def _offset(self) -> int:
-        operation = self.operation_factory.build_query_offset()
-        row = await self.submit_query_and_fetchone(operation)
+        operation = self.database_operation_factory.build_query_offset()
+        row = await self.execute_on_database_and_fetch_one(operation)
         return row[0] or 0

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

@@ -87,8 +87,8 @@ def __init__(
         self._transaction_repository = transaction_repository
 
     async def _setup(self) -> None:
-        operation = self.operation_factory.build_create()
-        await self.submit_query(operation)
+        operation = self.database_operation_factory.build_create()
+        await self.execute_on_database(operation)
 
     async def _destroy(self) -> None:
         await super()._destroy()
@@ -110,11 +110,11 @@ async def _find_entries(
         else:
             transaction_uuids = await transaction.uuids
 
-        operation = self.operation_factory.build_query(
+        operation = self.database_operation_factory.build_query(
             name, condition, ordering, limit, transaction_uuids, exclude_deleted
         )
 
-        async for row in self.submit_query_and_iter(operation, streaming_mode=streaming_mode):
+        async for row in self.execute_on_database_and_fetch_all(operation, streaming_mode=streaming_mode):
             yield SnapshotEntry(*row)
 
     async def is_synced(self, name: str, **kwargs) -> bool:
@@ -148,17 +148,17 @@ async def _synchronize(self, **kwargs) -> None:
         await self._store_offset(offset)
 
     async def _load_offset(self) -> int:
-        operation = self.operation_factory.build_query_offset()
+        operation = self.database_operation_factory.build_query_offset()
         # noinspection PyBroadException
         try:
-            row = await self.submit_query_and_fetchone(operation)
+            row = await self.execute_on_database_and_fetch_one(operation)
         except ProgrammingException:
             return 0
         return row[0]
 
     async def _store_offset(self, offset: int) -> None:
-        operation = self.operation_factory.build_submit_offset(offset)
-        await self.submit_query(operation)
+        operation = self.database_operation_factory.build_submit_offset(offset)
+        await self.execute_on_database(operation)
 
     async def _dispatch_one(self, event_entry: EventEntry, **kwargs) -> SnapshotEntry:
         if event_entry.action.is_delete:
@@ -210,8 +210,8 @@ async def _select_one_instance(self, name: str, uuid: UUID, **kwargs) -> RootEnt
         return snapshot_entry.build(**kwargs)
 
     async def _submit_entry(self, snapshot_entry: SnapshotEntry) -> SnapshotEntry:
-        operation = self.operation_factory.build_submit(**snapshot_entry.as_raw())
-        response = await self.submit_query_and_fetchone(operation)
+        operation = self.database_operation_factory.build_submit(**snapshot_entry.as_raw())
+        response = await self.execute_on_database_and_fetch_one(operation)
 
         snapshot_entry.created_at, snapshot_entry.updated_at = response
 
@@ -223,5 +223,5 @@ async def _clean_transactions(self, offset: int, **kwargs) -> None:
         )
         transaction_uuids = {transaction.uuid async for transaction in iterable}
         if len(transaction_uuids):
-            operation = self.operation_factory.build_delete(transaction_uuids)
-            await self.submit_query(operation)
+            operation = self.database_operation_factory.build_delete(transaction_uuids)
+            await self.execute_on_database(operation)

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

@@ -35,16 +35,16 @@ def __init__(self, *args, database_key: Optional[tuple[str]] = None, **kwargs):
         super().__init__(*args, database_key=database_key, **kwargs)
 
     async def _setup(self):
-        operation = self.operation_factory.build_create()
-        await self.submit_query(operation)
+        operation = self.database_operation_factory.build_create()
+        await self.execute_on_database(operation)
 
     async def _submit(self, transaction: TransactionEntry) -> TransactionEntry:
-        operation = self.operation_factory.build_submit(
+        operation = self.database_operation_factory.build_submit(
             **transaction.as_raw(),
         )
 
         try:
-            updated_at = await self.submit_query_and_fetchone(operation)
+            updated_at = await self.execute_on_database_and_fetch_one(operation)
         except ProgrammingException:
             raise TransactionRepositoryConflictException(
                 f"{transaction!r} status is invalid respect to the previous one."
@@ -53,6 +53,6 @@ async def _submit(self, transaction: TransactionEntry) -> TransactionEntry:
         return transaction
 
     async def _select(self, streaming_mode: Optional[bool] = None, **kwargs) -> AsyncIterator[TransactionEntry]:
-        operation = self.operation_factory.build_query(**kwargs)
-        async for row in self.submit_query_and_iter(operation, streaming_mode=streaming_mode):
+        operation = self.database_operation_factory.build_query(**kwargs)
+        async for row in self.execute_on_database_and_fetch_all(operation, streaming_mode=streaming_mode):
             yield TransactionEntry(*row, transaction_repository=self)

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

@@ -79,7 +79,7 @@ def _get_pool_from_factory(pool_factory: PoolFactory, database_key: Optional[tup
         return pool_factory.get_pool(type_="database")
 
     @property
-    def operation_factory(self) -> Optional[GenericDatabaseOperationFactory]:
+    def database_operation_factory(self) -> Optional[GenericDatabaseOperationFactory]:
         """Get the operation factory if any.
 
         :return: A ``OperationFactory`` if it has been set or ``None`` otherwise.
@@ -101,8 +101,8 @@ def _get_generic_operation_factory(self) -> Optional[type[GenericDatabaseOperati
                 raise TypeError(f"{type(self)!r} must contain a {DatabaseOperationFactory!r} as generic value.")
         return operation_factory_cls
 
-    async def submit_query_and_fetchone(self, operation: DatabaseOperation) -> tuple:
-        """Submit a SQL query and gets the first response.
+    async def execute_on_database_and_fetch_one(self, operation: DatabaseOperation) -> tuple:
+        """Submit an Operation and get the first response.
 
         :param operation: The operation to be executed.
         :return: This method does not return anything.
@@ -112,10 +112,10 @@ async def submit_query_and_fetchone(self, operation: DatabaseOperation) -> tuple
             return await client.fetch_one()
 
     # noinspection PyUnusedLocal
-    async def submit_query_and_iter(
+    async def execute_on_database_and_fetch_all(
         self, operation: DatabaseOperation, streaming_mode: Optional[bool] = None
     ) -> AsyncIterator[tuple]:
-        """Submit a SQL query and return an asynchronous iterator.
+        """Submit an Operation and return an asynchronous iterator.
 
         :param operation: The operation to be executed.
         :param streaming_mode: If ``True`` return the values in streaming directly from the database (keep an open
@@ -139,8 +139,8 @@ async def submit_query_and_iter(
             yield value
 
     # noinspection PyUnusedLocal
-    async def submit_query(self, operation: DatabaseOperation) -> None:
-        """Submit a SQL query.
+    async def execute_on_database(self, operation: DatabaseOperation) -> None:
+        """Submit an Operation.
 
         :param operation: The operation to be executed.
         :return: This method does not return anything.

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

@@ -52,22 +52,22 @@ async def test_pool(self):
     async def test_operation_factory(self):
         operation_factory = MockedLockDatabaseOperationFactory()
         mixin = DatabaseMixin(operation_factory=operation_factory)
-        self.assertEqual(operation_factory, mixin.operation_factory)
+        self.assertEqual(operation_factory, mixin.database_operation_factory)
 
     async def test_operation_factory_from_cls_init(self):
         mixin = DatabaseMixin(operation_factory_cls=LockDatabaseOperationFactory)
-        self.assertIsInstance(mixin.operation_factory, MockedLockDatabaseOperationFactory)
+        self.assertIsInstance(mixin.database_operation_factory, MockedLockDatabaseOperationFactory)
 
     async def test_operation_factory_from_cls_generic(self):
         class _DatabaseMixin(DatabaseMixin[LockDatabaseOperationFactory]):
             """For testing purposes."""
 
         mixin = _DatabaseMixin()
-        self.assertIsInstance(mixin.operation_factory, MockedLockDatabaseOperationFactory)
+        self.assertIsInstance(mixin.database_operation_factory, MockedLockDatabaseOperationFactory)
 
     async def test_operation_factory_none(self):
         mixin = DatabaseMixin()
-        self.assertEqual(None, mixin.operation_factory)
+        self.assertEqual(None, mixin.database_operation_factory)
 
     async def test_operation_factory_from_cls_generic_raises(self):
         class _DatabaseMixin(DatabaseMixin[int]):
@@ -76,76 +76,76 @@ class _DatabaseMixin(DatabaseMixin[int]):
         with self.assertRaises(TypeError):
             _DatabaseMixin()
 
-    async def test_submit_query(self):
+    async def test_execute_on_database(self):
         op1 = MockedDatabaseOperation("create_table")
         op2 = MockedDatabaseOperation("check_exist", [(True,)])
 
         async with DatabaseMixin() as database:
-            await database.submit_query(op1)
+            await database.execute_on_database(op1)
 
         async with MockedDatabaseClient(**self.config.get_default_database()) as client:
             await client.execute(op2)
             self.assertTrue((await client.fetch_one())[0])
 
-    async def test_submit_query_locked(self):
+    async def test_execute_on_database_locked(self):
         op1 = MockedDatabaseOperation("create_table", lock=1234)
         op2 = MockedDatabaseOperation("check_exist", [(True,)])
 
         async with DatabaseMixin() as database:
-            await database.submit_query(op1)
+            await database.execute_on_database(op1)
 
         async with MockedDatabaseClient(**self.config.get_default_database()) as client:
             await client.execute(op2)
             self.assertTrue((await client.fetch_one())[0])
 
-    async def test_submit_query_and_fetchone(self):
+    async def test_execute_on_database_and_fetch_one(self):
         op1 = MockedDatabaseOperation("create_table")
         op2 = MockedDatabaseOperation("insert")
         op3 = MockedDatabaseOperation("select", [(3,), (4,), (5,)])
 
         async with DatabaseMixin() as database:
-            await database.submit_query(op1)
-            await database.submit_query(op2)
+            await database.execute_on_database(op1)
+            await database.execute_on_database(op2)
 
-            observed = await database.submit_query_and_fetchone(op3)
+            observed = await database.execute_on_database_and_fetch_one(op3)
 
         self.assertEqual((3,), observed)
 
-    async def test_submit_query_and_iter(self):
+    async def test_execute_on_database_and_fetch_all(self):
         op1 = MockedDatabaseOperation("create_table")
         op2 = MockedDatabaseOperation("insert")
         op3 = MockedDatabaseOperation("select", [(3,), (4,), (5,)])
 
         async with DatabaseMixin() as database:
-            await database.submit_query(op1)
-            await database.submit_query(op2)
-            observed = [v async for v in database.submit_query_and_iter(op3)]
+            await database.execute_on_database(op1)
+            await database.execute_on_database(op2)
+            observed = [v async for v in database.execute_on_database_and_fetch_all(op3)]
 
         self.assertEqual([(3,), (4,), (5,)], observed)
 
-    async def test_submit_query_and_iter_streaming_mode_true(self):
+    async def test_execute_on_database_and_fetch_all_streaming_mode_true(self):
         op1 = MockedDatabaseOperation("create_table")
         op2 = MockedDatabaseOperation("insert")
         op3 = MockedDatabaseOperation("select", [(3,), (4,), (5,)])
 
         async with DatabaseMixin() as database:
-            await database.submit_query(op1)
-            await database.submit_query(op2)
+            await database.execute_on_database(op1)
+            await database.execute_on_database(op2)
 
-            observed = [v async for v in database.submit_query_and_iter(op3, streaming_mode=True)]
+            observed = [v async for v in database.execute_on_database_and_fetch_all(op3, streaming_mode=True)]
 
         self.assertEqual([(3,), (4,), (5,)], observed)
 
-    async def test_submit_query_and_iter_locked(self):
+    async def test_execute_on_database_and_fetch_all_locked(self):
         op1 = MockedDatabaseOperation("create_table", lock=1234)
         op2 = MockedDatabaseOperation("insert")
         op3 = MockedDatabaseOperation("select", [(3,), (4,), (5,)])
 
         async with DatabaseMixin() as database:
-            await database.submit_query(op1)
-            await database.submit_query(op2)
+            await database.execute_on_database(op1)
+            await database.execute_on_database(op2)
 
-            observed = [v async for v in database.submit_query_and_iter(op3)]
+            observed = [v async for v in database.execute_on_database_and_fetch_all(op3)]
 
         self.assertEqual([(3,), (4,), (5,)], observed)
 

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

@@ -120,8 +120,8 @@ async def _destroy(self) -> None:
         await super()._destroy()
 
     async def _create_table(self) -> None:
-        operation = self.operation_factory.build_create()
-        await self.submit_query(operation)
+        operation = self.database_operation_factory.build_create()
+        await self.execute_on_database(operation)
 
     async def _start_run(self) -> None:
         if self._run_task is None:
@@ -142,13 +142,13 @@ async def _flush_queue(self):
                 entry = self._queue.get_nowait()
             except QueueEmpty:
                 break
-            operation = self.operation_factory.build_mark_processed(entry.id_)
-            await self.submit_query(operation)
+            operation = self.database_operation_factory.build_mark_processed(entry.id_)
+            await self.execute_on_database(operation)
             self._queue.task_done()
 
     async def _enqueue(self, message: BrokerMessage) -> None:
-        operation = self.operation_factory.build_submit(message.topic, message.avro_bytes)
-        await self.submit_query(operation)
+        operation = self.database_operation_factory.build_submit(message.topic, message.avro_bytes)
+        await self.execute_on_database(operation)
         await self._notify_enqueued(message)
 
     # noinspection PyUnusedLocal
@@ -166,12 +166,12 @@ async def _dequeue(self) -> BrokerMessage:
                     logger.warning(
                         f"There was a problem while trying to deserialize the entry with {entry.id_!r} id: {exc}"
                     )
-                    operation = self.operation_factory.build_mark_processed(entry.id_)
-                    await self.submit_query(operation)
+                    operation = self.database_operation_factory.build_mark_processed(entry.id_)
+                    await self.execute_on_database(operation)
                     continue
 
-                operation = self.operation_factory.build_delete(entry.id_)
-                await self.submit_query(operation)
+                operation = self.database_operation_factory.build_delete(entry.id_)
+                await self.execute_on_database(operation)
                 return message
             finally:
                 self._queue.task_done()
@@ -195,8 +195,8 @@ async def _wait_enqueued(self) -> None:
 
     async def _get_count(self) -> int:
         # noinspection PyTypeChecker
-        operation = self.operation_factory.build_count(self.retry)
-        row = await self.submit_query_and_fetchone(operation)
+        operation = self.database_operation_factory.build_count(self.retry)
+        row = await self.execute_on_database_and_fetch_one(operation)
         count = row[0]
         return count
 
@@ -210,14 +210,14 @@ async def _dequeue_batch(self) -> None:
             entries = [_Entry(*row) for row in rows]
 
             ids = tuple(entry.id_ for entry in entries)
-            operation = self.operation_factory.build_mark_processing(ids)
+            operation = self.database_operation_factory.build_mark_processing(ids)
             await client.execute(operation)
 
         for entry in entries:
             await self._queue.put(entry)
 
     async def _dequeue_rows(self, client: DatabaseClient) -> list[Any]:
-        operation = self.operation_factory.build_query(self._retry, self._records)
+        operation = self.database_operation_factory.build_query(self._retry, self._records)
         await client.execute(operation)
         return [row async for row in client.fetch_all()]
 

packages/core/minos-microservice-networks/minos/networks/brokers/subscribers/filtered/validators/duplicates/database/impl.py

@@ -40,13 +40,13 @@ async def _setup(self) -> None:
         await self._create_table()
 
     async def _create_table(self) -> None:
-        operation = self.operation_factory.build_create()
-        await self.submit_query(operation)
+        operation = self.database_operation_factory.build_create()
+        await self.execute_on_database(operation)
 
     async def _is_unique(self, topic: str, uuid: UUID) -> bool:
-        operation = self.operation_factory.build_submit(topic, uuid)
+        operation = self.database_operation_factory.build_submit(topic, uuid)
         try:
-            await self.submit_query(operation)
+            await self.execute_on_database(operation)
             return True
         except IntegrityException:
             return False