Method batch_execute

src/server/tasks/dbbench/interaction.py:285–299

(self, sql: List[Union[str, Tuple[str, Union[Sequence, Dict[str, Any]]]]])

            return error_msg

    async def batch_execute(self, sql: List[Union[str, Tuple[str, Union[Sequence, Dict[str, Any]]]]]):
        final_sql = ''
        final_data = []
        for item in sql:
            if isinstance(item, str):
                query = item
                data = ()
            else:
                query, data = item
            if query.strip():
                if final_sql:
                    final_sql += ';\n'
                final_sql += query
                final_data.extend(data)
        await self.execute(final_sql, final_data)