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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
| import time
from mysql.connector.pooling import MySQLConnectionPool
class MySql:
def __init__(self, config):
self.expire = 1800
self.mysql_pool = MySQLConnectionPool(
host=config.MYSQL_HOST,
port=config.MYSQL_PORT,
user=config.MYSQL_USER,
passwd=config.MYSQL_PASSWORD,
database=config.MYSQL_DATABASE,
ssl_disabled=config.SSL_DISABLED,
pool_size=config.MYSQL_MAX_CONNECTIONS
)
def __get_connection(self):
try:
conn = self.mysql_pool.get_connection()
return conn
except:
time.sleep(5)
def select(self, sql: str, arg=None):
if not arg == None and not isinstance(arg, tuple):
arg = (arg,)
with self.__get_connection() as conn:
with conn.cursor() as cur:
cur.execute(sql, arg)
result = cur.fetchall()
return result
def insert(self, sql: str, arg=None):
if not arg == None and not isinstance(arg, tuple):
arg = (arg,)
with self.__get_connection() as conn:
with conn.cursor() as cur:
try:
cur.execute(sql, arg)
conn.commit()
return True
except Exception as e:
print(e)
conn.rollback()
return False
def update(self, sql: str, arg=None):
if not arg == None and not isinstance(arg, tuple):
arg = (arg,)
with self.__get_connection() as conn:
with conn.cursor() as cur:
try:
cur.execute(sql, arg)
conn.commit()
return True
except Exception as e:
print(e)
conn.rollback()
return False
def alter(self, sql: str, arg=None):
if not arg == None and not isinstance(arg, tuple):
arg = (arg,)
with self.__get_connection() as conn:
with conn.cursor() as cur:
try:
cur.execute(sql, arg)
conn.commit()
return True
except Exception as e:
print(e)
conn.rollback()
return False
def get_index(self, table: str):
sql = """
show
INDEX
from
""" + table + """
where
not Key_name = 'PRIMARY'
"""
with self.__get_connection() as conn:
with conn.cursor() as cur:
try:
cur.execute(sql)
result = cur.fetchall()
index_names = []
for i in result:
index_names.append(i[2])
return index_names
except Exception as e:
print(e)
conn.rollback()
return []
def truncate(self, sql: str, arg=None):
if not arg == None and not isinstance(arg, tuple):
arg = (arg,)
with self.__get_connection() as conn:
with conn.cursor() as cur:
try:
cur.execute(sql, arg)
conn.commit()
return True
except Exception as e:
print(e)
conn.rollback()
return False
|