use execute_many from backends directly

Author: Micheal Gendy

databases/backends/aiopg.py

@@ -168,7 +168,9 @@ async def execute(self, query: ClauseElement) -> typing.Any:
         finally:
             cursor.close()
 
-    async def execute_many(self, queries: typing.List[ClauseElement]) -> None:
+    async def execute_many(
+        self, queries: typing.List[ClauseElement], values: typing.List[dict]
+    ) -> None:
         assert self._connection is not None, "Connection is not acquired"
         cursor = await self._connection.cursor()
         try:

databases/backends/asyncmy.py

@@ -158,7 +158,9 @@ async def execute(self, query: ClauseElement) -> typing.Any:
             finally:
                 await cursor.close()
 
-    async def execute_many(self, queries: typing.List[ClauseElement]) -> None:
+    async def execute_many(
+        self, queries: typing.List[ClauseElement], values: typing.List[dict]
+    ) -> None:
         assert self._connection is not None, "Connection is not acquired"
         async with self._connection.cursor() as cursor:
             try:

databases/backends/mysql.py

@@ -158,13 +158,14 @@ async def execute(self, query: ClauseElement) -> typing.Any:
         finally:
             await cursor.close()
 
-    async def execute_many(self, queries: typing.List[ClauseElement]) -> None:
+    async def execute_many(
+        self, queries: typing.List[ClauseElement], values: typing.List[dict]
+    ) -> None:
         assert self._connection is not None, "Connection is not acquired"
         cursor = await self._connection.cursor()
+        query_str, values = self._compile_many(queries, values)
         try:
-            for single_query in queries:
-                single_query, args, context = self._compile(single_query)
-                await cursor.execute(single_query, args)
+            await cursor.executemany(query_str, values)
         finally:
             await cursor.close()
 
@@ -220,6 +221,18 @@ def _compile(
         logger.debug("Query: %s Args: %s", query_message, repr(args), extra=LOG_EXTRA)
         return compiled.string, args, CompilationContext(execution_context)
 
+    def _compile_many(
+        self, queries: typing.List[ClauseElement], values: typing.List[dict]
+    ) -> typing.Tuple[str, list]:
+        compiled = queries[0].compile(
+            dialect=self._dialect, compile_kwargs={"render_postcompile": True}
+        )
+        for args in values:
+            for key, val in args.items():
+                if key in compiled._bind_processors:
+                    args[key] = compiled._bind_processors[key](val)
+        return compiled.string, values
+
     @property
     def raw_connection(self) -> aiomysql.connection.Connection:
         assert self._connection is not None, "Connection is not acquired"

databases/backends/postgres.py

@@ -1,6 +1,5 @@
 import logging
 import typing
-from collections.abc import Sequence
 
 import asyncpg
 from sqlalchemy.dialects.postgresql import pypostgresql
@@ -217,14 +216,12 @@ async def execute(self, query: ClauseElement) -> typing.Any:
         query_str, args, result_columns = self._compile(query)
         return await self._connection.fetchval(query_str, *args)
 
-    async def execute_many(self, queries: typing.List[ClauseElement]) -> None:
+    async def execute_many(
+        self, queries: typing.List[ClauseElement], values: typing.List[dict]
+    ) -> None:
         assert self._connection is not None, "Connection is not acquired"
-        # asyncpg uses prepared statements under the hood, so we just
-        # loop through multiple executes here, which should all end up
-        # using the same prepared statement.
-        for single_query in queries:
-            single_query, args, result_columns = self._compile(single_query)
-            await self._connection.execute(single_query, *args)
+        query_str, values = self._compile_many(queries, values)
+        await self._connection.executemany(query_str, values)
 
     async def iterate(
         self, query: ClauseElement
@@ -269,6 +266,18 @@ def _compile(self, query: ClauseElement) -> typing.Tuple[str, list, tuple]:
         )
         return compiled_query, args, result_map
 
+    def _compile_many(
+        self, queries: typing.List[ClauseElement], values: typing.List[dict]
+    ) -> typing.Tuple[str, list]:
+        compiled = queries[0].compile(
+            dialect=self._dialect, compile_kwargs={"render_postcompile": True}
+        )
+        for args in values:
+            for key, val in args.items():
+                if key in compiled._bind_processors:
+                    args[key] = compiled._bind_processors[key](val)
+        return compiled.string, values
+
     @staticmethod
     def _create_column_maps(
         result_columns: tuple,

databases/backends/sqlite.py

@@ -135,10 +135,13 @@ async def execute(self, query: ClauseElement) -> typing.Any:
                 return cursor.rowcount
             return cursor.lastrowid
 
-    async def execute_many(self, queries: typing.List[ClauseElement]) -> None:
+    async def execute_many(
+        self, queries: typing.List[ClauseElement], values: typing.List[dict]
+    ) -> None:
         assert self._connection is not None, "Connection is not acquired"
-        for single_query in queries:
-            await self.execute(single_query)
+        query_str, values = self._compile_many(queries, values)
+        async with self._connection.cursor() as cursor:
+            await cursor.executemany(query_str, values)
 
     async def iterate(
         self, query: ClauseElement
@@ -194,6 +197,18 @@ def _compile(
         )
         return compiled.string, args, CompilationContext(execution_context)
 
+    def _compile_many(
+        self, queries: typing.List[ClauseElement], values: typing.List[dict]
+    ) -> typing.Tuple[str, list]:
+        compiled = queries[0].compile(
+            dialect=self._dialect, compile_kwargs={"render_postcompile": True}
+        )
+        for args in values:
+            for key, val in args.items():
+                if key in compiled._bind_processors:
+                    args[key] = compiled._bind_processors[key](val)
+        return compiled.string, values
+
     @property
     def raw_connection(self) -> aiosqlite.core.Connection:
         assert self._connection is not None, "Connection is not acquired"

databases/core.py

@@ -304,7 +304,7 @@ async def execute_many(
     ) -> None:
         queries = [self._build_query(query, values_set) for values_set in values]
         async with self._query_lock:
-            await self._connection.execute_many(queries)
+            await self._connection.execute_many(queries, values)
 
     async def iterate(
         self, query: typing.Union[ClauseElement, str], values: dict = None

databases/interfaces.py

@@ -37,7 +37,9 @@ async def fetch_val(
     async def execute(self, query: ClauseElement) -> typing.Any:
         raise NotImplementedError()  # pragma: no cover
 
-    async def execute_many(self, queries: typing.List[ClauseElement]) -> None:
+    async def execute_many(
+        self, queries: typing.List[ClauseElement], values: typing.List[dict]
+    ) -> None:
         raise NotImplementedError()  # pragma: no cover
 
     async def iterate(