(self, query: str, values: list)
| 194 | |
| 195 | @translate_exceptions |
| 196 | async def execute_many(self, query: str, values: list) -> None: |
| 197 | async with self.acquire_connection() as connection: |
| 198 | self.log.debug("%s: %s", query, values) |
| 199 | async with connection.cursor() as cursor: |
| 200 | if self.capabilities.supports_transactions: |
| 201 | await connection.begin() |
| 202 | try: |
| 203 | await cursor.executemany(query, values) |
| 204 | except Exception: |
| 205 | await connection.rollback() |
| 206 | raise |
| 207 | else: |
| 208 | await connection.commit() |
| 209 | else: |
| 210 | await cursor.executemany(query, values) |
| 211 | |
| 212 | @translate_exceptions |
| 213 | async def execute_query(self, query: str, values: list | None = None) -> tuple[int, list[dict]]: |
nothing calls this directly
no test coverage detected