| Total Complexity | 4 |
| Total Lines | 48 |
| Duplicated Lines | 0 % |
| Coverage | 73.32% |
| Changes | 0 | ||
| 1 | 1 | from typing import Dict, Optional, Union |
|
| 2 | |||
| 3 | 1 | from mysql.connector import MySQLConnection |
|
| 4 | |||
| 5 | 1 | from pystratum_mysql.MySqlConnector import MySqlConnector |
|
| 6 | |||
| 7 | |||
| 8 | 1 | class MySqlDefaultConnector(MySqlConnector): |
|
| 9 | """ |
||
| 10 | Connects to a MySQL instance using user name and password. |
||
| 11 | """ |
||
| 12 | |||
| 13 | # ------------------------------------------------------------------------------------------------------------------ |
||
| 14 | 1 | def __init__(self, params: Dict[str, Union[str, int]]): |
|
| 15 | """ |
||
| 16 | Object constructor. |
||
| 17 | |||
| 18 | :param params: The connection parameters. |
||
| 19 | """ |
||
| 20 | |||
| 21 | 1 | self._params: Dict[str, Union[str, int]] = params |
|
| 22 | """ |
||
| 23 | The connection parameters. |
||
| 24 | """ |
||
| 25 | |||
| 26 | 1 | self._connection: Optional[MySQLConnection] = None |
|
| 27 | 1 | """ |
|
| 28 | The connection between Python and the MySQL instance. |
||
| 29 | """ |
||
| 30 | |||
| 31 | # ------------------------------------------------------------------------------------------------------------------ |
||
| 32 | 1 | def connect(self) -> MySQLConnection: |
|
| 33 | """ |
||
| 34 | Connects to the MySQL instance. |
||
| 35 | """ |
||
| 36 | 1 | self._connection = MySQLConnection(**self._params) |
|
| 37 | |||
| 38 | return self._connection |
||
| 39 | |||
| 40 | # ------------------------------------------------------------------------------------------------------------------ |
||
| 41 | 1 | def disconnect(self) -> None: |
|
| 42 | """ |
||
| 43 | Disconnects from the MySQL instance. |
||
| 44 | """ |
||
| 45 | if self._connection: |
||
| 46 | self._connection.disconnect() |
||
| 47 | self._connection = None |
||
| 48 | |||
| 50 |