(self, query: str)
| 81 | |
| 82 | @translate_exceptions |
| 83 | async def execute_script(self, query: str) -> None: |
| 84 | async with self.acquire_connection() as connection: |
| 85 | self.log.debug(query) |
| 86 | async with connection.cursor() as cursor: |
| 87 | for q in query.split(";"): |
| 88 | if not q.strip(): |
| 89 | continue |
| 90 | await cursor.execute(q) |
| 91 | |
| 92 | @translate_exceptions |
| 93 | async def execute_insert(self, query: str, values: list) -> int: |
no test coverage detected