diff options
| author | Pinapelz <yukais@pinapelz.com> | 2023-11-17 13:24:42 -0800 |
|---|---|---|
| committer | Pinapelz <yukais@pinapelz.com> | 2023-11-17 13:24:42 -0800 |
| commit | 96019367e8f72eac26abd3b7a908c2b914bd1ae1 (patch) | |
| tree | fa7dbe66ea8b12bde0970dc8db27f7601464b3b8 /sql | |
| parent | c159e86a3487dbae79de8d0887f288625084bfcd (diff) | |
v2: add initial Next JS files, remove static templates
Diffstat (limited to 'sql')
| -rw-r--r-- | sql/sql_handler.py | 162 |
1 files changed, 0 insertions, 162 deletions
diff --git a/sql/sql_handler.py b/sql/sql_handler.py deleted file mode 100644 index 9d1c10d..0000000 --- a/sql/sql_handler.py +++ /dev/null @@ -1,162 +0,0 @@ -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): - 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) - - 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) -
\ No newline at end of file |
