add feature to ask user if not exist in insert data in scylla adaptor

Author: hanieh-mav

archipy/adapters/scylladb/adapters.py

@@ -382,18 +382,23 @@ def drop_table(self, table: str) -> None:
             raise
 
     @override
-    def insert(self, table: str, data: dict[str, Any], ttl: int | None = None) -> None:
+    def insert(self, table: str, data: dict[str, Any], ttl: int | None = None, if_not_exists: bool = False) -> None:
         """Insert data into a table.
 
         Args:
             table (str): The name of the table.
             data (dict[str, Any]): Key-value pairs representing column names and values.
             ttl (int | None): Time to live in seconds. If None, data persists indefinitely.
+            if_not_exists (bool): If True, use lightweight transaction (INSERT ... IF NOT EXISTS).
+                              This prevents errors on duplicate primary keys but is slow
         """
         columns = ", ".join(data.keys())
         placeholders = ", ".join(["%s" for _ in data.keys()])
         query = f"INSERT INTO {table} ({columns}) VALUES ({placeholders})"
 
+        if if_not_exists:
+            query += " IF NOT EXISTS"
+
         if ttl is not None:
             query += f" USING TTL {ttl}"
 
@@ -961,18 +966,29 @@ async def drop_table(self, table: str) -> None:
             raise
 
     @override
-    async def insert(self, table: str, data: dict[str, Any], ttl: int | None = None) -> None:
+    async def insert(
+        self,
+        table: str,
+        data: dict[str, Any],
+        ttl: int | None = None,
+        if_not_exists: bool = False,
+    ) -> None:
         """Insert data into a table asynchronously.
 
         Args:
             table (str): The name of the table.
             data (dict[str, Any]): Key-value pairs representing column names and values.
             ttl (int | None): Time to live in seconds. If None, data persists indefinitely.
+            if_not_exists (bool): If True, use lightweight transaction (INSERT ... IF NOT EXISTS).
+                              This prevents errors on duplicate primary keys but is slow
         """
         columns = ", ".join(data.keys())
         placeholders = ", ".join(["%s" for _ in data.keys()])
         query = f"INSERT INTO {table} ({columns}) VALUES ({placeholders})"
 
+        if if_not_exists:
+            query += " IF NOT EXISTS"
+
         if ttl is not None:
             query += f" USING TTL {ttl}"
 

archipy/adapters/scylladb/ports.py

@@ -129,13 +129,15 @@ def drop_table(self, table: str) -> None:
         raise NotImplementedError
 
     @abstractmethod
-    def insert(self, table: str, data: dict[str, Any], ttl: int | None = None) -> None:
+    def insert(self, table: str, data: dict[str, Any], ttl: int | None = None, if_not_exists: bool = False) -> None:
         """Insert data into a table.
 
         Args:
             table (str): The name of the table.
             data (dict[str, Any]): Key-value pairs representing column names and values.
             ttl (int | None): Time to live in seconds. If None, data persists indefinitely.
+            if_not_exists (bool): If True, use lightweight transaction (INSERT ... IF NOT EXISTS).
+                              This prevents errors on duplicate primary keys but is slow
 
         Raises:
             NotImplementedError: If not implemented by the subclass.
@@ -405,13 +407,21 @@ async def drop_table(self, table: str) -> None:
         raise NotImplementedError
 
     @abstractmethod
-    async def insert(self, table: str, data: dict[str, Any], ttl: int | None = None) -> None:
+    async def insert(
+        self,
+        table: str,
+        data: dict[str, Any],
+        ttl: int | None = None,
+        if_not_exists: bool = False,
+    ) -> None:
         """Insert data into a table asynchronously.
 
         Args:
             table (str): The name of the table.
             data (dict[str, Any]): Key-value pairs representing column names and values.
             ttl (int | None): Time to live in seconds. If None, data persists indefinitely.
+            if_not_exists (bool): If True, use lightweight transaction (INSERT ... IF NOT EXISTS).
+                              This prevents errors on duplicate primary keys but is slow
 
         Raises:
             NotImplementedError: If not implemented by the subclass.

features/scylladb_adapter.feature

@@ -217,3 +217,25 @@ Feature: ScyllaDB Adapter
     Given a ScyllaDB adapter is configured
     When I get pool statistics
     Then the pool statistics should be returned
+
+  @needs-scylladb
+  Scenario: Insert with If Not Exists
+    Given a ScyllaDB adapter is configured
+    And a keyspace "test_ks" with replication factor 1 exists
+    And a table "inventory" with schema "CREATE TABLE IF NOT EXISTS inventory (id int PRIMARY KEY, item text, quantity int)"
+    And data exists in table "inventory":
+      | id | item   | quantity |
+      | 1  | Widget | 100      |
+    When I insert data into table "inventory" with id 1, item "some_item", quantity 500
+    Then the table "inventory" should contain 1 row
+
+  @needs-scylladb @async
+  Scenario: Async Insert with If Not Exists
+    Given an async ScyllaDB adapter is configured
+    And an async keyspace "test_ks" with replication factor 1 exists
+    And an async table "async_inventory" with schema "CREATE TABLE IF NOT EXISTS async_inventory (id int PRIMARY KEY, item text, quantity int)"
+    And data exists in table "async_inventory":
+      | id | item   | quantity |
+      | 1  | Widget | 100      |
+    When I async insert data into table "async_inventory" with id 1, item "some_item", quantity 500
+    Then the async table "async_inventory" should contain 1 row

features/steps/scylladb_adapter_steps.py

@@ -991,3 +991,40 @@ def step_verify_pool_stats(context: Context) -> None:
     assert isinstance(stats, dict), "Pool statistics should be a dictionary"
     assert "monitoring_enabled" in stats, "Pool statistics should contain 'monitoring_enabled' key"
     logger.info("Pool statistics verified")
+
+
+# If not Exist in insert steps
+
+
+@when('I insert data into table "{table}" with id {id:d}, item "{item}", quantity {quantity:d}')
+def step_insert_with_if_not_exists(
+    context: Context, table: str, id: int, item: str, quantity: int,
+) -> None:
+    """Insert data in a table with if not exists.
+
+    Args:
+        context (Context): Behave context.
+        table (str): Table name.
+        id (int): User ID.
+        item (str): Item.
+        quantity (int): Quantity
+    """
+    adapter = get_scylladb_adapter(context)
+    adapter.insert(table=table, data={"id": id, "item": item, "quantity": quantity}, if_not_exists=True)
+    logger.info("Inserted user data into '%s'", table)
+
+
+@when('I async insert data into table "{table}" with id {id:d}, item "{item}", quantity {quantity:d}')
+async def step_insert_with_if_not_exists(context: Context, table: str, id: int, item: str, quantity: int) -> None:
+    """Insert data in a table with if not exists asynchronously.
+
+    Args:
+        context (Context): Behave context.
+        table (str): Table name.
+        id (int): User ID.
+        item (str): Item.
+        quantity (int): Quantity
+    """
+    adapter = get_scylladb_adapter(context)
+    await adapter.insert(table=table, data={"id": id, "item": item, "quantity": quantity}, if_not_exists=True)
+    logger.info("Inserted user data into '%s'", table)