Spaces:
Runtime error
Runtime error
| # sudo mysql -u root -p | |
| # CREATE DATABASE monitor_log; | |
| # USE monitor_log; | |
| # SHOW TABLES; | |
| # DROP TABLE anonymous_no; | |
| # DROP TABLE log; | |
| # SELECT * FROM information_schema.columns WHERE table_name = 'log'; | |
| import mysql.connector | |
| from mysql.connector import Error | |
| class Database: | |
| def __init__(self, host, user, password, database): | |
| """Initialize the database connection.""" | |
| self.host = host | |
| self.user = user | |
| self.password = password | |
| self.database = database | |
| self.connection = None | |
| def connect(self): | |
| """Establish a connection to the database.""" | |
| try: | |
| self.connection = mysql.connector.connect( | |
| host=self.host, | |
| user=self.user, | |
| password=self.password, | |
| database=self.database | |
| ) | |
| if self.connection.is_connected(): | |
| print("Successfully connected to the database.") | |
| except Error as e: | |
| print(f"Error: {e}") | |
| def disconnect(self): | |
| """Close the database connection.""" | |
| if self.connection.is_connected(): | |
| self.connection.close() | |
| print("Database connection closed.") | |
| def execute_query(self, query, values=None): | |
| """Execute a query that does not return results (INSERT, UPDATE, DELETE).""" | |
| cursor = self.connection.cursor() | |
| try: | |
| if values: | |
| cursor.execute(query, values) | |
| else: | |
| cursor.execute(query) | |
| self.connection.commit() | |
| print("Query executed successfully.") | |
| except Error as e: | |
| print(f"Error: {e}") | |
| finally: | |
| cursor.close() | |
| def fetch_all(self, query): | |
| """Execute a SELECT query and return all results.""" | |
| cursor = self.connection.cursor() | |
| results = [] | |
| try: | |
| cursor.execute(query) | |
| results = cursor.fetchall() | |
| return results | |
| except Error as e: | |
| print(f"Error: {e}") | |
| finally: | |
| cursor.close() | |
| def create_table(self, create_table_query): | |
| """Create a new table in the database.""" | |
| cursor = self.connection.cursor() | |
| try: | |
| cursor.execute(create_table_query) | |
| print("Table created successfully.") | |
| except Error as e: | |
| print(f"Error: {e}") | |
| finally: | |
| cursor.close() | |
| if __name__ == "__main__": | |
| db = Database(host='localhost', user='root', password='0123456789', database='monitor_log') | |
| db.connect() | |
| db.create_table(""" | |
| CREATE TABLE anonymous_no ( | |
| number INT AUTO_INCREMENT PRIMARY KEY, | |
| name VARCHAR(60) NOT NULL, | |
| description VARCHAR(500) | |
| ); | |
| """) | |
| db.create_table(""" | |
| CREATE TABLE log ( | |
| id VARCHAR(255) AUTO_INCREMENT PRIMARY KEY, | |
| name VARCHAR(60) NOT NULL, | |
| datetime DATETIME DEFAULT CURRENT_TIMESTAMP | |
| ); | |
| """) | |
| db.disconnect() |