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 |