diff options
Diffstat (limited to 'sql/sql_handler.py')
| -rw-r--r-- | sql/sql_handler.py | 76 |
1 files changed, 49 insertions, 27 deletions
diff --git a/sql/sql_handler.py b/sql/sql_handler.py index d349f64..a02db52 100644 --- a/sql/sql_handler.py +++ b/sql/sql_handler.py @@ -12,19 +12,18 @@ class SQLHandler: 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, exclude=None) -> mysql.connector: + 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 - ) + 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( @@ -48,75 +47,77 @@ class SQLHandler: print(err) exit(1) - def create_table(self, table_name: str, table_columns: str): + def create_table(self, name: str, column: str): cursor = self.connection.cursor() try: - cursor.execute(f"CREATE TABLE {table_name} ({table_columns})") - print(f"Table {table_name} created successfully") + cursor.execute(f"CREATE TABLE {name} ({column})") + print(f"Table {name} created successfully") except Error as err: print(err) - def insert_data(self, table_name: str, table_columns: str, data: str): + def insert_row(self, name: str, column: str, data: tuple): cursor = self.connection.cursor() try: - cursor.execute( - f"INSERT INTO {table_name} ({table_columns}) VALUES ({data})") + placeholders = ', '.join(['%s'] * len(data)) + query = f"INSERT INTO {name} ({column}) VALUES ({placeholders})" + cursor.execute(query, data) self.connection.commit() - print("Data inserted successfully") + print("Data Inserted:", data) except Error as err: print("Error inserting data") print(err) - def clear_table(self, table_name: str): + + def clear_table(self, name: str): cursor = self.connection.cursor() try: - cursor.execute(f"DELETE FROM {table_name}") + 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, table_name: str): + def reset_auto_increment(self, name: str): cursor = self.connection.cursor() try: - cursor.execute(f"ALTER TABLE {table_name} AUTO_INCREMENT = 1") + 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, table_name: str, new_table_name: str, table_columns: str): + def copy_rows_to_new_table(self, name: str, new_name: str, column: str): cursor = self.connection.cursor() try: cursor.execute( - f"INSERT INTO {new_table_name} ({table_columns}) SELECT {table_columns} FROM {table_name}") + f"INSERT INTO {new_name} ({column}) SELECT {column} FROM {name}") cursor.execute( - f"ALTER TABLE {new_table_name} MODIFY COLUMN id INT AUTO_INCREMENT") + 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, table_name: str): + def drop_table(self, name: str): cursor = self.connection.cursor() try: - cursor.execute(f"DROP TABLE {table_name}") + 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, table_name: str, column_name: str, value: str): + def check_row_exists(self, name: str, column_name: str, value: str): """ Checks if a row exists in a table """ cursor = self.connection.cursor() try: - cursor.execute(f"SELECT * FROM {table_name} WHERE {column_name} = '{value}'") + cursor.execute(f"SELECT * FROM {name} WHERE {column_name} = '{value}'") result = cursor.fetchone() if result: return True @@ -126,15 +127,36 @@ class SQLHandler: print("Error checking row") print(err) - def update_row(self, table_name: str, column_name: str, search_val: str, replace_col:str, new_value: str): + 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() try: - cursor.execute(f"UPDATE {table_name} SET {replace_col} = '{new_value}' WHERE {column_name} = '{search_val}'") + 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)
\ No newline at end of file + print(err) + + def execute_query(self, query: str): + cursor = self.connection.cursor() + 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() + try: + cursor.execute(query) + result = cursor.fetchall() + return result + except Error as err: + print("Error executing query") + print(err) +
\ No newline at end of file |