yiekheng
d73439698a
- move Python sources into app package and switch services to module entrypoints - relocate Dockerfiles under docker/, add buildx publish script, override compose for local builds - configure images to pull from gitea.04080616.xyz/yiekheng with env-driven tags and limits - harden installs and transfer worker logging/concurrency for cleaner container output
99 lines
3.1 KiB
Python
99 lines
3.1 KiB
Python
import mysql.connector
|
|
from mysql.connector import Error
|
|
|
|
class DB:
|
|
def __init__(self):
|
|
self.config = {
|
|
'host': '192.168.0.210',
|
|
'user': 'rex_cm',
|
|
'password': 'hengserver',
|
|
'database': 'rex_cm',
|
|
'port': 3306
|
|
}
|
|
self.init_database()
|
|
|
|
def get_connection(self):
|
|
"""Get MySQL database connection."""
|
|
try:
|
|
connection = mysql.connector.connect(**self.config)
|
|
return connection
|
|
except Error as e:
|
|
print(f"Error connecting to MySQL: {e}")
|
|
return None
|
|
|
|
def init_database(self):
|
|
"""Initialize the database connection."""
|
|
connection = self.get_connection()
|
|
if connection is None:
|
|
raise Exception("Failed to connect to database")
|
|
|
|
try:
|
|
cursor = connection.cursor()
|
|
# Test connection by checking if required tables exist
|
|
cursor.execute("SHOW TABLES LIKE 'acc'")
|
|
if not cursor.fetchone():
|
|
raise Exception("Table 'acc' does not exist")
|
|
|
|
cursor.execute("SHOW TABLES LIKE 'user'")
|
|
if not cursor.fetchone():
|
|
raise Exception("Table 'user' does not exist")
|
|
|
|
# print("Database connection verified - required tables exist")
|
|
|
|
except Error as e:
|
|
print(f"Error verifying database: {e}")
|
|
raise Exception(f"Database verification failed: {e}")
|
|
finally:
|
|
if connection.is_connected():
|
|
cursor.close()
|
|
connection.close()
|
|
|
|
def query(self, query, params=None):
|
|
"""Execute a query and return results."""
|
|
connection = self.get_connection()
|
|
if connection is None:
|
|
return []
|
|
|
|
try:
|
|
cursor = connection.cursor(dictionary=True)
|
|
|
|
if params:
|
|
cursor.execute(query, params)
|
|
else:
|
|
cursor.execute(query)
|
|
|
|
results = cursor.fetchall()
|
|
return results
|
|
|
|
except Error as e:
|
|
print(f"Error executing query: {e}")
|
|
return []
|
|
finally:
|
|
if connection.is_connected():
|
|
cursor.close()
|
|
connection.close()
|
|
|
|
def execute(self, query, params=None):
|
|
"""Execute a query that modifies data (INSERT, UPDATE, DELETE) and return success status."""
|
|
connection = self.get_connection()
|
|
if connection is None:
|
|
return False
|
|
|
|
try:
|
|
cursor = connection.cursor()
|
|
|
|
if params:
|
|
cursor.execute(query, params)
|
|
else:
|
|
cursor.execute(query)
|
|
|
|
connection.commit()
|
|
return True
|
|
|
|
except Error as e:
|
|
print(f"Error executing query: {e}")
|
|
return False
|
|
finally:
|
|
if connection.is_connected():
|
|
cursor.close()
|
|
connection.close() |