(self, sql: List[Union[str, Tuple[str, Union[Sequence, Dict[str, Any]]]]])
| 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() |
| 109 | try: |
| 110 | async with await conn.cursor() as cursor: |
| 111 | for item in sql: |
| 112 | if isinstance(item, str): |
| 113 | query = item |
| 114 | data = () |
| 115 | else: |
| 116 | query, data = item |
| 117 | if query.strip(): |
| 118 | await cursor.execute(query, data) |
| 119 | await conn.commit() |
| 120 | except Exception as e: |
| 121 | self.logger.error(f"MySQL Error during batch execution", exc_info=e) |
| 122 | try: |
| 123 | await conn.rollback() |
| 124 | except: |
| 125 | self.logger.exception(f"Rollback failed") |
| 126 | raise e |
| 127 | |
| 128 | async def _create_database(self): |
| 129 | database = f'dbbench_{self.session_id.replace("-", "_")}' |
no test coverage detected