"""
MySQL Database Module
Handles all database connections and operations
"""
import mysql.connector
from mysql.connector import Error
import json
from pathlib import Path
from datetime import datetime, timedelta
from decimal import Decimal
def load_config():
"""Load database configuration from config.json"""
try:
config_path = Path(__file__).parent.parent / 'config.json'
with open(config_path, 'r') as f:
config = json.load(f)
return config.get('mysql', {})
except Exception as error:
raise Exception(f"Failed to load config: {str(error)}")
def get_connection():
"""Create and return a MySQL database connection"""
try:
config = load_config()
connection = mysql.connector.connect(
host=config.get('host', 'localhost'),
port=config.get('port', 3306),
user=config.get('user', 'root'),
password=config.get('password', ''),
database=config.get('database', 'subscription_tracker')
)
return connection
except Error as error:
raise Exception(f"Database connection failed: {str(error)}")
def execute_query(query, params=None, fetch_one=False, fetch_all=False):
"""
Execute a SQL query with parameters
Args:
query: SQL query string
params: Query parameters (tuple or dict)
fetch_one: Return single row
fetch_all: Return all rows
"""
connection = None
cursor = None
try:
connection = get_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute(query, params or ())
if fetch_one:
result = cursor.fetchone()
return convert_decimals(result) if result else None
elif fetch_all:
results = cursor.fetchall()
return [convert_decimals(row) for row in results]
else:
connection.commit()
return {
'success': True,
'affected_rows': cursor.rowcount,
'last_insert_id': cursor.lastrowid
}
except Error as error:
if connection:
connection.rollback()
raise Exception(f"Query execution failed: {str(error)}")
finally:
if cursor:
cursor.close()
if connection:
connection.close()
def convert_decimals(data):
"""Convert Decimal objects to float for JSON serialization"""
if isinstance(data, dict):
return {key: convert_decimals(value) for key, value in data.items()}
elif isinstance(data, list):
return [convert_decimals(item) for item in data]
elif isinstance(data, Decimal):
return float(data)
elif isinstance(data, (datetime, )):
return data.isoformat()
return data
def test_connection():
"""Test database connection"""
try:
connection = get_connection()
cursor = connection.cursor()
cursor.execute("SELECT 1")
result = cursor.fetchone()
cursor.close()
connection.close()
return {'success': True, 'message': 'Database connection successful'}
except Exception as error:
return {'success': False, 'error': str(error)}
