(self, query: str, values: list | None = None)
| 211 | |
| 212 | @translate_exceptions |
| 213 | async def execute_query(self, query: str, values: list | None = None) -> tuple[int, list[dict]]: |
| 214 | async with self.acquire_connection() as connection: |
| 215 | self.log.debug("%s: %s", query, values) |
| 216 | async with connection.cursor() as cursor: |
| 217 | await cursor.execute(query, values) |
| 218 | rows = await cursor.fetchall() |
| 219 | if rows: |
| 220 | fields = [f.name for f in cursor._result.fields] |
| 221 | return cursor.rowcount, [dict(zip(fields, row)) for row in rows] |
| 222 | return cursor.rowcount, [] |
| 223 | |
| 224 | async def execute_query_dict(self, query: str, values: list | None = None) -> list[dict]: |
| 225 | return (await self.execute_query(query, values))[1] |
no test coverage detected