(
self, query: str, values: list | None = None
)
| 147 | |
| 148 | @translate_exceptions |
| 149 | async def execute_query( |
| 150 | self, query: str, values: list | None = None |
| 151 | ) -> tuple[int, Sequence[dict]]: |
| 152 | query = query.replace("\x00", "'||CHAR(0)||'") |
| 153 | async with self.acquire_connection() as connection: |
| 154 | self.log.debug("%s: %s", query, values) |
| 155 | start = connection.total_changes |
| 156 | rows = await connection.execute_fetchall(query, values) |
| 157 | return (connection.total_changes - start) or len(rows), rows |
| 158 | |
| 159 | @translate_exceptions |
| 160 | async def execute_query_dict(self, query: str, values: list | None = None) -> list[dict]: |
no test coverage detected