Spaces:
Runtime error
Runtime error
File size: 2,951 Bytes
ef9b1b9 | 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 | # 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() |