feat: use DML batches in executemany() method

google/cloud/spanner_dbapi/cursor.py

@@ -258,9 +258,26 @@ def executemany(self, operation, seq_of_params):
 
         many_result_set = StreamedManyResultSets()
 
-        for params in seq_of_params:
-            self.execute(operation, params)
-            many_result_set.add_iter(self._itr)
+        if classification in (parse_utils.STMT_INSERT, parse_utils.STMT_UPDATING):
+            statements = []
+
+            for params in seq_of_params:
+                sql, params = parse_utils.sql_pyformat_args_to_spanner(
+                    operation, params
+                )
+                statements.append((sql, params, get_param_types(params)))
+
+            transaction = self.connection.transaction_checkout()
+            _, res = transaction.batch_update(statements)
+
+            if self.connection.autocommit:
+                transaction.commit()
+
+            many_result_set.add_iter(res)
+        else:
+            for params in seq_of_params:
+                self.execute(operation, params)
+                many_result_set.add_iter(self._itr)
 
         self._result_set = many_result_set
         self._itr = many_result_set

tests/unit/spanner_dbapi/test_cursor.py

@@ -337,6 +337,42 @@ def test_executemany(self):
             (mock.call(operation, (1,)), mock.call(operation, (2,)))
         )
 
+    def test_executemany_insert_batch(self):
+        from google.cloud.spanner_v1.param_types import INT64
+        from google.cloud.spanner_dbapi import connect
+
+        sql = """INSERT INTO table (col1, "col2", `col3`, `"col4"`) VALUES (%s, %s, %s, %s)"""
+
+        with mock.patch(
+            "google.cloud.spanner_v1.instance.Instance.exists", return_value=True
+        ):
+            with mock.patch(
+                "google.cloud.spanner_v1.database.Database.exists", return_value=True,
+            ):
+                connection = connect("test-instance", "test-database")
+
+        cursor = connection.cursor()
+
+        connection._transaction = mock.Mock(committed=False, rolled_back=False)
+        connection._transaction.batch_update = mock.Mock(return_value=[None, []])
+
+        cursor.executemany(sql, [(1, 2, 3, 4), (5, 6, 7, 8)])
+
+        connection._transaction.batch_update.assert_called_once_with(
+            [
+                (
+                    """INSERT INTO table (col1, "col2", `col3`, `"col4"`) VALUES (@a0, @a1, @a2, @a3)""",
+                    {"a0": 1, "a1": 2, "a2": 3, "a3": 4},
+                    {"a0": INT64, "a1": INT64, "a2": INT64, "a3": INT64},
+                ),
+                (
+                    """INSERT INTO table (col1, "col2", `col3`, `"col4"`) VALUES (@a0, @a1, @a2, @a3)""",
+                    {"a0": 5, "a1": 6, "a2": 7, "a3": 8},
+                    {"a0": INT64, "a1": INT64, "a2": INT64, "a3": INT64},
+                ),
+            ]
+        )
+
     @unittest.skipIf(
         sys.version_info[0] < 3, "Python 2 has an outdated iterator definition"
     )