ISSUE #98

* Increase coverage.
* Minor improvements.

Author: Sergio García Prado

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

@@ -2,6 +2,7 @@
     ABC,
     abstractmethod,
 )
+from collections.abc import Iterable
 from datetime import (
     datetime,
 )
@@ -35,7 +36,7 @@ def build_create_table(self) -> DatabaseOperation:
     @abstractmethod
     def build_submit_row(
         self,
-        transaction_uuids: tuple[UUID],
+        transaction_uuids: Iterable[UUID],
         uuid: UUID,
         action: Action,
         name: str,

packages/plugins/minos-database-aiopg/minos/plugins/aiopg/factories/aggregate/events.py

@@ -15,6 +15,7 @@
 from psycopg2.sql import (
     SQL,
     Composable,
+    Identifier,
     Literal,
     Placeholder,
 )
@@ -40,6 +41,13 @@
 class AiopgEventDatabaseOperationFactory(EventDatabaseOperationFactory):
     """Aiopg Event Database Operation Factory class."""
 
+    def build_table_name(self) -> str:
+        """Get the table name.
+
+        :return: A ``str`` value.
+        """
+        return "aggregate_event"
+
     def build_create_table(self) -> DatabaseOperation:
         """Build the database operation to create the event table.
 
@@ -68,11 +76,11 @@ def build_create_table(self) -> DatabaseOperation:
                     $$
                     LANGUAGE plpgsql;
                     """,
-                    lock="aggregate_event",
+                    lock=self.build_table_name(),
                 ),
                 AiopgDatabaseOperation(
-                    """
-                    CREATE TABLE IF NOT EXISTS aggregate_event (
+                    f"""
+                    CREATE TABLE IF NOT EXISTS {self.build_table_name()} (
                         id BIGSERIAL PRIMARY KEY,
                         action ACTION_TYPE NOT NULL,
                         uuid UUID NOT NULL,
@@ -84,14 +92,14 @@ def build_create_table(self) -> DatabaseOperation:
                         UNIQUE (uuid, version, transaction_uuid)
                     );
                     """,
-                    lock="aggregate_event",
+                    lock=self.build_table_name(),
                 ),
             ]
         )
 
     def build_submit_row(
         self,
-        transaction_uuids: tuple[UUID],
+        transaction_uuids: Iterable[UUID],
         uuid: UUID,
         action: Action,
         name: str,
@@ -118,7 +126,7 @@ def build_submit_row(
         """
         insert_values = SQL(
             """
-            INSERT INTO aggregate_event (id, action, uuid, name, version, data, created_at, transaction_uuid)
+            INSERT INTO {table_name} (id, action, uuid, name, version, data, created_at, transaction_uuid)
             VALUES (
                 default,
                 %(action)s,
@@ -151,7 +159,7 @@ def build_submit_row(
 
         from_sql, from_parameters = self._build_submit_from(transaction_uuids)
 
-        query = insert_values.format(from_parts=from_sql)
+        query = insert_values.format(from_parts=from_sql, table_name=Identifier(self.build_table_name()))
         parameters = from_parameters | insert_parameters
 
         return AiopgDatabaseOperation(query, parameters, lock)
@@ -160,7 +168,7 @@ def _build_submit_from(self, transaction_uuids: Iterable[UUID]) -> tuple[Composa
         select_transaction = SQL(
             """
             SELECT {index} AS transaction_index, uuid, MAX(version) AS version
-            FROM aggregate_event
+            FROM {table_name}
             WHERE uuid = %(uuid)s AND transaction_uuid = {transaction_uuid}
             GROUP BY uuid
             """
@@ -171,7 +179,13 @@ def _build_submit_from(self, transaction_uuids: Iterable[UUID]) -> tuple[Composa
             name = f"transaction_uuid_{index}"
             parameters[name] = transaction_uuid
 
-            from_query_parts.append(select_transaction.format(index=Literal(index), transaction_uuid=Placeholder(name)))
+            from_query_parts.append(
+                select_transaction.format(
+                    index=Literal(index),
+                    transaction_uuid=Placeholder(name),
+                    table_name=Identifier(self.build_table_name()),
+                ),
+            )
 
         query = SQL(" UNION ALL ").join(from_query_parts)
         return query, parameters
@@ -193,7 +207,7 @@ def build_select_rows(
         id_ge: Optional[int] = None,
         transaction_uuid: Optional[UUID] = None,
         transaction_uuid_ne: Optional[UUID] = None,
-        transaction_uuid_in: Optional[tuple[UUID, ...]] = None,
+        transaction_uuid_in: Optional[Iterable[UUID, ...]] = None,
         **kwargs,
     ) -> DatabaseOperation:
         """Build the database operation to select rows.
@@ -216,10 +230,12 @@ def build_select_rows(
 
         :return: A ``DatabaseOperation`` instance.
         """
+        if transaction_uuid_in is not None:
+            transaction_uuid_in = tuple(transaction_uuid_in)
 
-        _select_all = """
+        _select_all = f"""
             SELECT uuid, name, version, data, id, action, created_at, transaction_uuid
-            FROM aggregate_event
+            FROM {self.build_table_name()}
             """
 
         conditions = list()
@@ -284,7 +300,7 @@ def build_select_max_id(self) -> DatabaseOperation:
 
         :return: A ``DatabaseOperation`` instance.
         """
-        return AiopgDatabaseOperation("SELECT MAX(id) FROM aggregate_event;".strip())
+        return AiopgDatabaseOperation(f"SELECT MAX(id) FROM {self.build_table_name()};".strip())
 
 
 AiopgDatabaseClient.register_factory(EventDatabaseOperationFactory, AiopgEventDatabaseOperationFactory)

packages/plugins/minos-database-aiopg/minos/plugins/aiopg/factories/aggregate/snapshots/impl.py

@@ -39,6 +39,20 @@
 class AiopgSnapshotDatabaseOperationFactory(SnapshotDatabaseOperationFactory):
     """Aiopg Snapshot Database Operation Factory class."""
 
+    def build_table_name(self) -> str:
+        """Get the table name.
+
+        :return: A ``str`` value.
+        """
+        return "snapshot"
+
+    def build_offset_table_name(self) -> str:
+        """Get the offset table name.
+
+        :return: A ``str`` value.
+        """
+        return "snapshot_aux_offset"
+
     def build_create_table(self) -> DatabaseOperation:
         """Build the database operation to create the snapshot table.
 
@@ -51,8 +65,8 @@ def build_create_table(self) -> DatabaseOperation:
                     lock="uuid-ossp",
                 ),
                 AiopgDatabaseOperation(
-                    """
-                    CREATE TABLE IF NOT EXISTS snapshot (
+                    f"""
+                    CREATE TABLE IF NOT EXISTS {self.build_table_name()} (
                         uuid UUID NOT NULL,
                         name TEXT NOT NULL,
                         version INT NOT NULL,
@@ -64,17 +78,17 @@ def build_create_table(self) -> DatabaseOperation:
                         PRIMARY KEY (uuid, transaction_uuid)
                     );
                     """,
-                    lock="snapshot",
+                    lock=self.build_table_name(),
                 ),
                 AiopgDatabaseOperation(
-                    """
-                    CREATE TABLE IF NOT EXISTS snapshot_aux_offset (
+                    f"""
+                    CREATE TABLE IF NOT EXISTS {self.build_offset_table_name()} (
                         id bool PRIMARY KEY DEFAULT TRUE,
                         value BIGINT NOT NULL,
                         CONSTRAINT id_uni CHECK (id)
                     );
                     """,
-                    lock="snapshot_aux_offset",
+                    lock=self.build_offset_table_name(),
                 ),
             ]
         )
@@ -86,8 +100,8 @@ def build_delete_by_transactions(self, transaction_uuids: Iterable[UUID]) -> Dat
         :return: A ``DatabaseOperation`` instance.
         """
         return AiopgDatabaseOperation(
-            """
-            DELETE FROM snapshot
+            f"""
+            DELETE FROM {self.build_table_name()}
             WHERE transaction_uuid IN %(transaction_uuids)s;
             """,
             {"transaction_uuids": tuple(transaction_uuids)},
@@ -118,8 +132,10 @@ def build_insert(
         """
 
         return AiopgDatabaseOperation(
-            """
-            INSERT INTO snapshot (uuid, name, version, schema, data, created_at, updated_at, transaction_uuid)
+            f"""
+            INSERT INTO {self.build_table_name()} (
+                uuid, name, version, schema, data, created_at, updated_at, transaction_uuid
+            )
             VALUES (
                 %(uuid)s,
                 %(name)s,
@@ -153,7 +169,7 @@ def build_query(
         condition: _Condition,
         ordering: Optional[_Ordering],
         limit: Optional[int],
-        transaction_uuids: tuple[UUID, ...],
+        transaction_uuids: Iterable[UUID],
         exclude_deleted: bool,
     ) -> DatabaseOperation:
         """Build the query database operation.
@@ -172,7 +188,13 @@ def build_query(
         :return: A ``DatabaseOperation`` instance.
         """
         builder = AiopgSnapshotQueryDatabaseOperationBuilder(
-            name, condition, ordering, limit, transaction_uuids, exclude_deleted
+            name=name,
+            condition=condition,
+            ordering=ordering,
+            limit=limit,
+            transaction_uuids=transaction_uuids,
+            exclude_deleted=exclude_deleted,
+            table_name=self.build_table_name(),
         )
         query, parameters = builder.build()
 
@@ -185,14 +207,17 @@ def build_store_offset(self, value: int) -> DatabaseOperation:
         :return: A ``DatabaseOperation`` instance.
         """
         return AiopgDatabaseOperation(
-            """
-            INSERT INTO snapshot_aux_offset (id, value)
+            f"""
+            INSERT INTO {self.build_offset_table_name()} (id, value)
             VALUES (TRUE, %(value)s)
             ON CONFLICT (id)
-            DO UPDATE SET value = GREATEST(%(value)s, (SELECT value FROM snapshot_aux_offset WHERE id = TRUE));
+            DO UPDATE SET value = GREATEST(
+                %(value)s,
+                (SELECT value FROM {self.build_offset_table_name()} WHERE id = TRUE)
+            );
             """.strip(),
             {"value": value},
-            lock="insert_snapshot_aux_offset",
+            lock=f"insert_{self.build_offset_table_name()}",
         )
 
     def build_get_offset(self) -> DatabaseOperation:
@@ -201,9 +226,9 @@ def build_get_offset(self) -> DatabaseOperation:
         :return: A ``DatabaseOperation`` instance.
         """
         return AiopgDatabaseOperation(
-            """
+            f"""
             SELECT value
-            FROM snapshot_aux_offset
+            FROM {self.build_offset_table_name()}
             WHERE id = TRUE;
             """
         )

packages/plugins/minos-database-aiopg/minos/plugins/aiopg/factories/aggregate/snapshots/queries.py

@@ -4,6 +4,7 @@
 
 from typing import (
     Any,
+    Iterable,
     Optional,
 )
 from uuid import (
@@ -64,15 +65,21 @@ def __init__(
         condition: _Condition,
         ordering: Optional[_Ordering] = None,
         limit: Optional[int] = None,
-        transaction_uuids: tuple[UUID, ...] = (NULL_UUID,),
+        transaction_uuids: Iterable[UUID, ...] = (NULL_UUID,),
         exclude_deleted: bool = False,
+        table_name: Optional[str] = None,
     ):
+        if not isinstance(transaction_uuids, tuple):
+            transaction_uuids = tuple(transaction_uuids)
+        if table_name is None:
+            table_name = "snapshot"
         self.name = name
         self.condition = condition
         self.ordering = ordering
         self.limit = limit
         self.transaction_uuids = transaction_uuids
         self.exclude_deleted = exclude_deleted
+        self.table_name = table_name
         self._parameters = None
 
     def build(self) -> tuple[Composable, dict[str, Any]]:
@@ -116,7 +123,9 @@ def _build_select_from(self) -> Composable:
             self._parameters[name] = transaction_uuid
 
             from_query_parts.append(
-                self._SELECT_TRANSACTION_CHUNK.format(index=Literal(index), transaction_uuid=Placeholder(name))
+                self._SELECT_TRANSACTION_CHUNK.format(
+                    index=Literal(index), transaction_uuid=Placeholder(name), table_name=Identifier(self.table_name)
+                )
             )
 
         from_query = SQL(" UNION ALL ").join(from_query_parts)
@@ -264,7 +273,7 @@ def generate_random_str() -> str:
 
     _SELECT_TRANSACTION_CHUNK = SQL(
         "SELECT {index} AS transaction_index, * "
-        "FROM snapshot "
+        "FROM {table_name} "
         "WHERE name = %(name)s AND transaction_uuid = {transaction_uuid} "
     )