(self, query: str, values: list | None = None)
| 158 | |
| 159 | @translate_exceptions |
| 160 | async def execute_query_dict(self, query: str, values: list | None = None) -> list[dict]: |
| 161 | query = query.replace("\x00", "'||CHAR(0)||'") |
| 162 | async with self.acquire_connection() as connection: |
| 163 | self.log.debug("%s: %s", query, values) |
| 164 | return list(map(dict, await connection.execute_fetchall(query, values))) |
| 165 | |
| 166 | @translate_exceptions |
| 167 | async def execute_script(self, query: str) -> None: |
nothing calls this directly
no test coverage detected