Method create_connection

tortoise/backends/mysql/client.py:111–150

(self, with_db: bool)

        self._pool_init_lock = asyncio.Lock()

    async def create_connection(self, with_db: bool) -> None:
        if charset_by_name(self.charset) is None:
            raise DBConnectionError(f"Unknown charset {self.charset}")
        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 mysql.create_pool(password=self.password, **self._template)

            if isinstance(self._pool, mysql.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
                        # Only set session timezone when use_tz=True
                        if get_use_tz():
                            from datetime import datetime as _datetime

                            offset = _datetime.now(tz=get_default_timezone()).utcoffset()
                            total_seconds = int(offset.total_seconds())  # type: ignore
                            hours = total_seconds // 3600
                            minutes = abs(total_seconds) % 3600 // 60
                            tz = f"{hours:+d}:{minutes:02d}"
                            await cursor.execute(f"SET time_zone='{tz}';")
            await self._post_connect()
            self.log.debug("Created connection %s pool with params: %s", self._pool, self._template)
        except errors.OperationalError:
            raise DBConnectionError(f"Can't connect to MySQL server: {self._template}")

    async def _expire_connections(self) -> None:
        if self._pool:  # pragma: nobranch