Open ShichaoMa opened 4 years ago
import typing
from sqlalchemy.sql import ClauseElement
async def execute_many(self, queries: typing.List[ClauseElement]) -> None:
assert self._connection is not None, "Connection is not acquired"
cursor = await self._connection.cursor()
try:
prepared_queries = []
for single_query in queries:
single_query, args, context = self._compile(single_query)
prepared_queries.append((single_query, args))
await cursor.executemany(prepared_queries)
finally:
await cursor.close()
execute single sql many times?
aiomysql sames work better