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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
|
import mysql.connector
from mysql.connector import Error, errorcode
class SQLHandler:
def __init__(self, host_name: str, user_name: str, user_password: str, database_name: str):
self.host_name = host_name
self.username = user_name
self.password = user_password
self.database_name = database_name
self.connection = self._create_server_connection(
host_name, user_name, user_password)
self._load_database(database_name)
def _create_server_connection(self, host_name: str, user_name: str, user_password: str) -> mysql.connector:
connection = None
try:
connection = mysql.connector.connect(host=host_name, user=user_name, passwd=user_password)
print("MySQL Database connection successful")
except Error as err:
print(f"Error: '{err}'")
return connection
def get_connection(self):
return self.connection
def _create_database(self, cursor: str, database_name: str):
try:
cursor.execute(
f"CREATE DATABASE {database_name} DEFAULT CHARACTER SET 'utf8'")
except Error as err:
print(f"Failed creating database: {err}")
exit(1)
def _load_database(self, database_name: str):
cursor = self.connection.cursor(buffered=True)
try:
cursor.execute(f"USE {database_name}")
print(f"Database {database_name} loaded successfully")
except Error as err:
print(f"Database {database_name} does not exist")
if err.errno == errorcode.ER_BAD_DB_ERROR:
self._create_database(cursor, database_name)
print(f"Database {database_name} created successfully")
self.connection.database = database_name
else:
print(err)
exit(1)
def create_table(self, name: str, column: str):
cursor = self.connection.cursor(buffered=True)
try:
cursor.execute(f"CREATE TABLE {name} ({column})")
print(f"Table {name} created successfully")
except Error as err:
print(err)
def insert_row(self, name: str, column: str, data: tuple):
cursor = self.connection.cursor(buffered=True)
try:
placeholders = ', '.join(['%s'] * len(data))
query = f"INSERT INTO {name} ({column}) VALUES ({placeholders})"
cursor.execute(query, data)
self.connection.commit()
print("Data Inserted:", data, "into", name)
except Error as err:
print("Error inserting data")
print(err)
def clear_table(self, name: str):
cursor = self.connection.cursor(buffered=True)
try:
cursor.execute(f"DELETE FROM {name}")
self.connection.commit()
print("Table cleared successfully")
except Error as err:
print("Error clearing table")
print(err)
def reset_auto_increment(self, name: str):
cursor = self.connection.cursor(buffered=True)
try:
cursor.execute(f"ALTER TABLE {name} AUTO_INCREMENT = 1")
self.connection.commit()
print("Table reset successfully")
except Error as err:
print("Error resetting table")
print(err)
def copy_rows_to_new_table(self, name: str, new_name: str, column: str):
cursor = self.connection.cursor(buffered=True)
try:
cursor.execute(
f"INSERT INTO {new_name} ({column}) SELECT {column} FROM {name}")
cursor.execute(
f"ALTER TABLE {new_name} MODIFY COLUMN id INT AUTO_INCREMENT")
self.connection.commit()
print("Rows copied successfully")
except Error as err:
print("Error copying rows")
print(err)
def drop_table(self, name: str):
cursor = self.connection.cursor(buffered=True)
try:
cursor.execute(f"DROP TABLE {name}")
self.connection.commit()
print("Table dropped successfully")
except Error as err:
print("Error dropping table")
print(err)
def check_row_exists(self, name: str, column_name: str, value: str):
"""
Checks if a row exists in a table
"""
cursor = self.connection.cursor(buffered=True)
try:
cursor.execute(f"SELECT * FROM {name} WHERE {column_name} = '{value}'")
result = cursor.fetchone()
if result:
return True
else:
return False
except Error as err:
print("Error checking row")
print(err)
def update_row(self, name: str, column_name: str, search_val: str, replace_col:str, new_value: str):
"""
Updates a row in a table
"""
cursor = self.connection.cursor(buffered=True)
try:
cursor.execute(f"UPDATE {name} SET {replace_col} = '{new_value}' WHERE {column_name} = '{search_val}'")
self.connection.commit()
print("Row updated successfully")
except Error as err:
print("Error updating row")
print(err)
def execute_query(self, query: str, data: tuple = None):
cursor = self.connection.cursor(buffered=True)
if data:
try:
cursor.execute(query, data)
result = cursor.fetchall()
return result
except Error as err:
print("Error executing query")
print(err)
return None
try:
cursor.execute(query)
result = cursor.fetchall()
return result
except Error as err:
print("Error executing query")
print(err)
def get_query_result(self, query: str):
cursor = self.connection.cursor(buffered=True)
try:
cursor.execute(query)
result = cursor.fetchall()
return result
except Error as err:
print("Error executing query")
print(err)
|
