-
Notifications
You must be signed in to change notification settings - Fork 0
/
dbworker.py
104 lines (75 loc) · 2.89 KB
/
dbworker.py
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
import mysql.connector
from mysql.connector import errorcode
import config
def connection(function):
def wrapper(*args, **kwargs):
con, cursor, response = None, None, None
try:
con = mysql.connector.connect(**config.DB_PARAMS)
con.autocommit = True
if con.is_connected():
cursor = con.cursor(dictionary=True)
response = function(cursor, *args, **kwargs)
print('Функция {} успешно применена'.format(function))
except mysql.connector.Error as e:
print('Ошибка подключения к базе данных:', e)
finally:
if con is not None and con.is_connected():
cursor.close()
con.close()
return response
return wrapper
@connection
def new_user(cursor, pk: int, username: str):
try:
cursor.execute("INSERT INTO users (id, username, status) VALUES ({}, '{}', {})".format(
pk, username,
config.Status.S_START.value))
except mysql.connector.Error as e:
if e.errno == errorcode.ER_DUP_ENTRY:
pass
else:
print(e)
@connection
def get_status(cursor, pk: int):
cursor.execute("SELECT status FROM users WHERE id={}".format(pk))
result = cursor.fetchone()
if result:
return int(result['status'])
@connection
def set_status(cursor, pk: int, status: int):
cursor.execute("UPDATE users SET status={} WHERE id={}".format(status, pk))
@connection
def get_group(cursor, pk: int):
cursor.execute("SELECT uni_group FROM users WHERE id={}".format(pk))
result = cursor.fetchone()
if result:
response = result['uni_group']
return response
@connection
def set_group(cursor, pk: int, group: str):
cursor.execute("UPDATE users SET uni_group='{}' WHERE id={}".format(group, pk))
@connection
def get_group_key(cursor, group: str) -> str:
cursor.execute("SELECT api_key FROM uni_groups WHERE id='{}'".format(group))
return cursor.fetchone()['api_key']
@connection
def group_exists(cursor, group: str) -> bool:
cursor.execute("SELECT id FROM uni_groups WHERE id='{}'".format(group))
return cursor.fetchone() is not None
@connection
def get_all_groups(cursor) -> list:
cursor.execute("SELECT id FROM uni_groups")
return list(row['id'] for row in cursor.fetchall())
@connection
def is_signed_up(cursor, pk: int) -> bool:
cursor.execute("SELECT signed_up FROM users WHERE id={}".format(pk))
result = cursor.fetchone()
return bool(result['signed_up'])
@connection
def sign_up_or_off(cursor, pk: int, sign_status: bool):
cursor.execute("UPDATE users SET signed_up={} WHERE id={}".format(sign_status, pk))
@connection
def get_all_signed(cursor) -> list:
cursor.execute("SELECT id FROM users WHERE signed_up=1")
return list(row['id'] for row in cursor.fetchall())