text: import asyncio
  from functools import wraps
  from typing import List, Optional, SupportsInt, Union
  
  import aiomysql
  import pymysql
  from pypika import MySQLQuery
  
  from tortoise.backends.base.client import (
      BaseDBAsyncClient,
      BaseTransactionWrapper,
      Capabilities,
      ConnectionWrapper,
      NestedTransactionContext,
      PoolConnectionWrapper,
      TransactionContext,
      TransactionContextPooled,
  )
  from tortoise.backends.mysql.executor import MySQLExecutor
  from tortoise.backends.mysql.schema_generator import MySQLSchemaGenerator
  from tortoise.exceptions import (
      DBConnectionError,
      IntegrityError,
      OperationalError,
      TransactionManagementError,
  )
  
  
  def translate_exceptions(func):
      @wraps(func)
      async def translate_exceptions_(self, *args):
          try:
              return await func(self, *args)
          except (
              pymysql.err.OperationalError,
              pymysql.err.ProgrammingError,
              pymysql.err.DataError,
              pymysql.err.InternalError,
              pymysql.err.NotSupportedError,
          ) as exc:
              raise OperationalError(exc)
          except pymysql.err.IntegrityError as exc:
              raise IntegrityError(exc)
  
      return translate_exceptions_
  
  
  class MySQLClient(BaseDBAsyncClient):
      query_class = MySQLQuery
      executor_class = MySQLExecutor
      schema_generator = MySQLSchemaGenerator
      capabilities = Capabilities("mysql", requires_limit=True, inline_comment=True)
  
      def __init__(
          self, *, user: str, password: str, database: str, host: str, port: SupportsInt, **kwargs
      ) -> None:
          super().__init__(**kwargs)
  
          self.user = user
          self.password = password
          self.database = database
          self.host = host
          self.port = int(port)  # make sure port is int type
          self.extra = kwargs.copy()
          self.storage_engine = self.extra.pop("storage_engine", "")
          self.extra.pop("connection_name", None)
          self.extra.pop("fetch_inserted", None)
          self.extra.pop("db", None)
          self.extra.pop("autocommit", None)
          self.charset = self.extra.pop("charset", "")
          self.pool_minsize = int(self.extra.pop("minsize", 1))
          self.pool_maxsize = int(self.extra.pop("maxsize", 5))
  
          self._template: dict = {}
          self._pool: Optional[aiomysql.Pool] = None
          self._connection = None
  
      async def create_connection(self, with_db: bool) -> None:
          self._template = {
              "host": self.host,
              "port": self.port,
              "user": self.user,
              "db": self.database if with_db else None,
              "autocommit": True,
              "charset": self.charset,
              "minsize": self.pool_minsize,
              "maxsize": self.pool_maxsize,
              **self.extra,
          }
          try:
              self._pool = await aiomysql.create_pool(password=self.password, **self._template)
  
              if isinstance(self._pool, aiomysql.Pool):
                  async with self.acquire_connection() as connection:
                      async with connection.cursor() as cursor:
                          if self.storage_engine:
                              await cursor.execute(
                                  f"SET default_storage_engine='{self.storage_engine}';"
                              )
                              if self.storage_engine.lower() != "innodb":  # pragma: nobranch
                                  self.capabilities.__dict__["supports_transactions"] = False
  
              self.log.debug("Created connection %s pool with params: %s", self._pool, self._template)
          except pymysql.err.OperationalError:
              raise DBConnectionError(f"Can't connect to MySQL server: {self._template}")
  
      async def _expire_connections(self) -> None:
          if self._pool:  # pragma: nobranch
              for conn in self._pool._free:
                  conn._reader.set_exception(EOFError("EOF"))
  
      async def _close(self) -> None:
          if self._pool:  # pragma: nobranch
              self._pool.close()
              await self._pool.wait_closed()
              self.log.debug("Closed connection %s with params: %s", self._connection, self._template)
              self._pool = None
  
      async def close(self) -> None:
          await self._close()
          self._template.clear()
  
      async def db_create(self) -> None:
          await self.create_connection(with_db=False)
          await self.execute_script(f"CREATE DATABASE {self.database}")
          await self.close()
  
      async def db_delete(self) -> None:
          await self.create_connection(with_db=False)
          try:
              await self.execute_script(f"DROP DATABASE {self.database}")
          except pymysql.err.DatabaseError:  # pragma: nocoverage
              pass
          await self.close()
  
      def acquire_connection(self) -> Union["ConnectionWrapper", "PoolConnectionWrapper"]:
          return PoolConnectionWrapper(self._pool)
  
      def _in_transaction(self) -> "TransactionContext":
          return TransactionContextPooled(TransactionWrapper(self))
  
      @translate_exceptions
      async def execute_insert(self, query: str, values: list) -> int:
          async with self.acquire_connection() as connection:
              self.log.debug("%s: %s", query, values)
              async with connection.cursor() as cursor:
                  await cursor.execute(query, values)
                  return cursor.lastrowid  # return auto-generated id
  
      @translate_exceptions
      async def execute_many(self, query: str, values: list) -> None:
          async with self.acquire_connection() as connection:
              self.log.debug("%s: %s", query, values)
              async with connection.cursor() as cursor:
                  if self.capabilities.supports_transactions:
                      await connection.begin()
                      try:
                          await cursor.executemany(query, values)
                      except Exception:
                          await connection.rollback()
                          raise
                      else:
                          await connection.commit()
                  else:
                      await cursor.executemany(query, values)
  
      @translate_exceptions
      async def execute_query(self, query: str, values: Optional[list] = None) -> List[dict]:
          async with self.acquire_connection() as connection:
              self.log.debug("%s: %s", query, values)
              async with connection.cursor() as cursor:
                  await cursor.execute(query, values)
                  rows = await cursor.fetchall()
                  if rows:
                      fields = [f.name for f in cursor._result.fields]
                      return [dict(zip(fields, row)) for row in rows]
                  return []
  
      async def execute_query_dict(self, query: str, values: Optional[list] = None) -> List[dict]:
          return await self.execute_query(query, values)
  
      @translate_exceptions
      async def execute_script(self, query: str) -> None:
          async with self.acquire_connection() as connection:
              self.log.debug(query)
              async with connection.cursor() as cursor:
                  await cursor.execute(query)
  
  
  class TransactionWrapper(MySQLClient, BaseTransactionWrapper):
      def __init__(self, connection) -> None:
          self.connection_name = connection.connection_name
          self._connection: aiomysql.Connection = connection._connection
          self._lock = asyncio.Lock()
          self.log = connection.log
          self._finalized: Optional[bool] = None
          self.fetch_inserted = connection.fetch_inserted
          self._parent = connection
  
      def _in_transaction(self) -> "TransactionContext":
          return NestedTransactionContext(self)
  
      def acquire_connection(self) -> ConnectionWrapper:
          return ConnectionWrapper(self._connection, self._lock)
  
      @translate_exceptions
      async def execute_many(self, query: str, values: list) -> None:
          async with self.acquire_connection() as connection:
              self.log.debug("%s: %s", query, values)
              async with connection.cursor() as cursor:
                  await cursor.executemany(query, values)
  
      @translate_exceptions
      async def start(self) -> None:
          await self._connection.begin()
          self._finalized = False
  
      async def commit(self) -> None:
          if self._finalized:
              raise TransactionManagementError("Transaction already finalised")
          await self._connection.commit()
          self._finalized = True
  
      async def rollback(self) -> None:
          if self._finalized:
              raise TransactionManagementError("Transaction already finalised")
          await self._connection.rollback()
          self._finalized = True
  