1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
|
import time
import logging
from abc import ABC, abstractmethod
from mysql.connector import connect, MySQLConnection, Error
from typing import Optional
from ..config import ConfigUnit
logger = logging.getLogger(__name__)
datetime_fmt = '%Y-%m-%d %H:%M:%S'
class MySQLCredsConfig(ConfigUnit, ABC):
@classmethod
def schema(cls) -> Optional[dict]:
schema = {}
for k in ('host', 'database', 'user', 'password'):
schema[k] = dict(type='string', required=True)
return schema
class MySQLHomeCredsConfig(MySQLCredsConfig):
NAME = 'mysql_home_creds'
class MySQLCloudCredsConfig(MySQLCredsConfig):
NAME = 'mysql_cloud_creds'
def mysql_now() -> str:
return time.strftime('%Y-%m-%d %H:%M:%S')
class MySQLDatabase(ABC):
_enable_pings: bool
_link: MySQLConnection
_time_zone: Optional[str]
@abstractmethod
def creds(self) -> MySQLCredsConfig:
pass
def __init__(self, enable_pings=False, time_zone='+01:00'):
self._enable_pings = enable_pings
self._time_zone = time_zone
self._connect()
def _connect(self):
c = self.creds()
self._link = connect(
host=c['host'],
user=c['user'],
password=c['password'],
database=c['database'],
)
if self._time_zone:
self._link.time_zone = self._time_zone
def cursor(self, **kwargs):
if self._enable_pings:
try:
self._link.ping(reconnect=True, attempts=2)
except Error as e:
logger.exception(e)
self._connect()
return self._link.cursor(**kwargs)
def commit(self):
self._link.commit()
class MySQLHomeDatabase(MySQLDatabase):
def creds(self) -> MySQLCredsConfig:
return MySQLHomeCredsConfig()
|
