(self, query: str, values: list | None = None)
| 129 | |
| 130 | @translate_exceptions |
| 131 | async def execute_query(self, query: str, values: list | None = None) -> tuple[int, list[dict]]: |
| 132 | async with self.acquire_connection() as connection: |
| 133 | self.log.debug("%s: %s", query, values) |
| 134 | if values: |
| 135 | params = [query, *values] |
| 136 | else: |
| 137 | params = [query] |
| 138 | normalized = query.lstrip().upper() |
| 139 | if ( |
| 140 | normalized.startswith("UPDATE") |
| 141 | or normalized.startswith("DELETE") |
| 142 | or normalized.startswith("INSERT") |
| 143 | ): |
| 144 | res = await connection.execute(*params) |
| 145 | try: |
| 146 | rows_affected = int(res.split(" ")[1]) |
| 147 | except Exception: # pragma: nocoverage |
| 148 | rows_affected = 0 |
| 149 | return rows_affected, [] |
| 150 | |
| 151 | rows = await connection.fetch(*params) |
| 152 | return len(rows), rows |
| 153 | |
| 154 | @translate_exceptions |
| 155 | async def execute_query_dict(self, query: str, values: list | None = None) -> list[dict]: |
nothing calls this directly
no test coverage detected