(self, sql: str, data: Union[Sequence, Dict[str, Any]] = ())
| 82 | self.container_ip = None |
| 83 | |
| 84 | async def execute(self, sql: str, data: Union[Sequence, Dict[str, Any]] = ()) -> str: |
| 85 | conn = await self._get_conn() |
| 86 | try: |
| 87 | async with await conn.cursor() as cursor: |
| 88 | results = [] |
| 89 | await cursor.execute(sql, data) |
| 90 | if cursor.with_rows: |
| 91 | rows = await cursor.fetchall() |
| 92 | results.extend(rows) |
| 93 | result_str = str(results) |
| 94 | await conn.commit() |
| 95 | except Exception as e: |
| 96 | result_str = str(e) |
| 97 | self.logger.error(f"MySQL Error during execution\nSQL: {sql}", exc_info=e) |
| 98 | try: |
| 99 | await conn.rollback() |
| 100 | except: |
| 101 | self.logger.exception(f"Rollback failed") |
| 102 | # Truncate |
| 103 | if len(result_str) > 800: |
| 104 | result_str = result_str[:800] + "[TRUNCATED]" |
| 105 | return result_str |
| 106 | |
| 107 | async def batch_execute(self, sql: List[Union[str, Tuple[str, Union[Sequence, Dict[str, Any]]]]]): |
| 108 | conn = await self._get_conn() |
no test coverage detected