Moved query building logic in Connection

Author: pvanliefland

databases/core.py

@@ -93,32 +93,31 @@ async def fetch_all(
         self, query: typing.Union[ClauseElement, str], values: dict = None
     ) -> typing.List[RowProxy]:
         async with self.connection() as connection:
-            return await connection.fetch_all(query=self._build_query(query, values))
+            return await connection.fetch_all(query, values)
 
     async def fetch_one(
         self, query: typing.Union[ClauseElement, str], values: dict = None
     ) -> RowProxy:
         async with self.connection() as connection:
-            return await connection.fetch_one(query=self._build_query(query, values))
+            return await connection.fetch_one(query, values)
 
     async def execute(
         self, query: typing.Union[ClauseElement, str], values: dict = None
     ) -> typing.Any:
         async with self.connection() as connection:
-            return await connection.execute(self._build_query(query, values))
+            return await connection.execute(query, values)
 
     async def execute_many(
         self, query: typing.Union[ClauseElement, str], values: list
     ) -> None:
         async with self.connection() as connection:
-            queries = [self._build_query(query, values_set) for values_set in values]
-            return await connection.execute_many(queries)
+            return await connection.execute_many(query, values)
 
     async def iterate(
         self, query: typing.Union[ClauseElement, str], values: dict = None
     ) -> typing.AsyncGenerator[RowProxy, None]:
         async with self.connection() as connection:
-            async for record in connection.iterate(self._build_query(query, values)):
+            async for record in connection.iterate(query, values):
                 yield record
 
     def connection(self) -> "Connection":
@@ -135,19 +134,6 @@ def connection(self) -> "Connection":
     def transaction(self, *, force_rollback: bool = False) -> "Transaction":
         return self.connection().transaction(force_rollback=force_rollback)
 
-    @staticmethod
-    def _build_query(
-        query: typing.Union[ClauseElement, str], values: dict = None
-    ) -> ClauseElement:
-        if isinstance(query, str):
-            query = text(query)
-
-            return query.bindparams(**values) if values is not None else query
-        elif values:
-            return query.values(**values)
-
-        return query
-
 
 class Connection:
     def __init__(self, backend: DatabaseBackend) -> None:
@@ -179,22 +165,31 @@ async def __aexit__(
             if self._connection_counter == 0:
                 await self._connection.release()
 
-    async def fetch_all(self, query: ClauseElement) -> typing.Any:
-        return await self._connection.fetch_all(query=query)
+    async def fetch_all(
+        self, query: typing.Union[ClauseElement, str], values: dict = None
+    ) -> typing.Any:
+        return await self._connection.fetch_all(self._build_query(query, values))
 
-    async def fetch_one(self, query: ClauseElement) -> typing.Any:
-        return await self._connection.fetch_one(query=query)
+    async def fetch_one(
+        self, query: typing.Union[ClauseElement, str], values: dict = None
+    ) -> typing.Any:
+        return await self._connection.fetch_one(self._build_query(query, values))
 
-    async def execute(self, query: ClauseElement) -> typing.Any:
-        return await self._connection.execute(query)
+    async def execute(
+        self, query: typing.Union[ClauseElement, str], values: dict = None
+    ) -> typing.Any:
+        return await self._connection.execute(self._build_query(query, values))
 
-    async def execute_many(self, queries: typing.List[ClauseElement]) -> None:
+    async def execute_many(
+        self, query: typing.Union[ClauseElement, str], values: list
+    ) -> None:
+        queries = [self._build_query(query, values_set) for values_set in values]
         await self._connection.execute_many(queries)
 
     async def iterate(
-        self, query: ClauseElement
+        self, query: typing.Union[ClauseElement, str], values: dict = None
     ) -> typing.AsyncGenerator[typing.Any, None]:
-        async for record in self._connection.iterate(query):
+        async for record in self._connection.iterate(self._build_query(query, values)):
             yield record
 
     def transaction(self, *, force_rollback: bool = False) -> "Transaction":
@@ -204,6 +199,19 @@ def transaction(self, *, force_rollback: bool = False) -> "Transaction":
     def raw_connection(self) -> typing.Any:
         return self._connection.raw_connection
 
+    @staticmethod
+    def _build_query(
+        query: typing.Union[ClauseElement, str], values: dict = None
+    ) -> ClauseElement:
+        if isinstance(query, str):
+            query = text(query)
+
+            return query.bindparams(**values) if values is not None else query
+        elif values:
+            return query.values(**values)
+
+        return query
+
 
 class Transaction:
     def __init__(self, connection: Connection, force_rollback: bool) -> None: