(self, query: str, values: list)
| 62 | |
| 63 | @translate_exceptions |
| 64 | async def execute_insert(self, query: str, values: list) -> int: |
| 65 | async with self.acquire_connection() as connection: |
| 66 | self.log.debug("%s: %s", query, values) |
| 67 | async with connection.cursor() as cursor: |
| 68 | await cursor.execute(f"SET NOCOUNT ON; {query}; SELECT @@IDENTITY", values) |
| 69 | return (await cursor.fetchone())[0] |
| 70 | |
| 71 | async def db_delete(self) -> None: |
| 72 | if not self.database: |
no test coverage detected