(
self,
*,
user: str,
password: str,
database: str,
host: str,
port: SupportsInt,
**kwargs: Any,
)
| 76 | ) |
| 77 | |
| 78 | def __init__( |
| 79 | self, |
| 80 | *, |
| 81 | user: str, |
| 82 | password: str, |
| 83 | database: str, |
| 84 | host: str, |
| 85 | port: SupportsInt, |
| 86 | **kwargs: Any, |
| 87 | ) -> None: |
| 88 | super().__init__(**kwargs) |
| 89 | |
| 90 | self.user = user |
| 91 | self.password = password |
| 92 | self.database = database |
| 93 | self.host = host |
| 94 | self.port = int(port) # make sure port is int type |
| 95 | self.extra = kwargs.copy() |
| 96 | self.storage_engine = self.extra.pop("storage_engine", "") |
| 97 | self.extra.pop("connection_name", None) |
| 98 | self.extra.pop("fetch_inserted", None) |
| 99 | self.extra.pop("db", None) |
| 100 | self.extra.pop("autocommit", None) |
| 101 | self.extra.setdefault("sql_mode", "STRICT_TRANS_TABLES") |
| 102 | self.charset = self.extra.pop("charset", "utf8mb4") |
| 103 | self.pool_minsize = int(self.extra.pop("minsize", 1)) |
| 104 | self.pool_maxsize = int(self.extra.pop("maxsize", 5)) |
| 105 | |
| 106 | self._template: dict = {} |
| 107 | self._pool: mysql.Pool | None = None |
| 108 | self._connection = None |
| 109 | self._pool_init_lock = asyncio.Lock() |
| 110 | |
| 111 | async def create_connection(self, with_db: bool) -> None: |
| 112 | if charset_by_name(self.charset) is None: |
nothing calls this directly
no outgoing calls
no test coverage detected