feat: support executemany

fakesnow/fakes.py

@@ -141,7 +141,7 @@ def _rewrite_params(
     ) -> str:
         if isinstance(params, dict):
             # see https://docs.snowflake.com/en/developer-guide/python-connector/python-connector-api
-            raise NotImplementedError("params dict not supported yet")
+            raise NotImplementedError("dict params not supported yet")
 
         if params and self._conn._paramstyle in ("pyformat", "format"):  # noqa: SLF001
             # duckdb uses question mark style params
@@ -239,6 +239,25 @@ def execute(
 
         return self
 
+    def executemany(
+        self,
+        command: str,
+        seqparams: Sequence[Any] | dict[str, Any],
+        **kwargs: Any,
+    ) -> FakeSnowflakeCursor:
+        if isinstance(seqparams, dict):
+            # see https://docs.snowflake.com/en/developer-guide/python-connector/python-connector-api
+            raise NotImplementedError("dict params not supported yet")
+
+        # TODO: support insert optimisations
+        # the snowflake connector will optimise inserts into a single query
+        # unless num_statements != 1 .. but for simplicity we execute each
+        # query one by one, which means the response differs
+        for p in seqparams:
+            self.execute(command, p)
+
+        return self
+
     def fetchall(self) -> list[tuple] | list[dict]:
         if self._use_dict_result:
             return self._duck_conn.fetch_arrow_table().to_pylist()

tests/test_fakes.py

@@ -215,6 +215,16 @@ def test_describe(cur: snowflake.connector.cursor.SnowflakeCursor):
     # fmt: on
 
 
+def test_executemany(cur: snowflake.connector.cursor.SnowflakeCursor):
+    cur.execute("create table customers (ID int, FIRST_NAME varchar, LAST_NAME varchar)")
+
+    customers = [(1, "Jenny", "P"), (2, "Jasper", "M")]
+    cur.executemany("insert into customers (id, first_name, last_name) values (%s,%s,%s)", customers)
+
+    cur.execute("select id, first_name, last_name from customers")
+    assert cur.fetchall() == customers
+
+
 def test_execute_string(conn: snowflake.connector.SnowflakeConnection):
     [_, cur2] = conn.execute_string(
         """ create table customers (ID int, FIRST_NAME varchar, LAST_NAME varchar);