Build a Clean, Efficient MySQL Utility Class in Python

Introduction

Database operations are a core part of daily DevOps work, and Python offers flexible ways to handle them. This guide explains how to design a concise, efficient MySQL utility class to simplify database tasks, improve code robustness, and enhance maintainability.

Design Philosophy

The class focuses on four key aspects:

Singleton pattern : Guarantees a single database connection instance across the application to avoid resource waste and boost performance.

Exception handling : Captures runtime errors and provides detailed logs for quick troubleshooting.

Logging : Records critical operations to aid auditing and fault diagnosis.

Simplicity & maintainability : Minimizes redundant code, improving readability and future maintenance.

Code Example

import traceback
import logging
from threading import Lock
from functools import lru_cache
import pymysql
from config_util import ConfigManager

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def singleton(cls):
    instances = {}
    lock = Lock()
    @lru_cache(maxsize=None)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            with lock:
                if cls not in instances:
                    instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance

@singleton
class MyMysql:
    def __init__(self):
        self._config_load()

    def _config_load(self):
        config = ConfigManager()
        self._mysql_ip = config.get_config_default('Mysql', 'IP', '127.0.0.1')
        self._mysql_port = int(config.get_config_default('Mysql', 'Port', '3306'))
        self._mysql_user = config.get_config_default('Mysql', 'User', 'root')
        self._mysql_pwd = config.get_config_default('Mysql', 'Password', '')
        self._mysql_db = config.get_config_default('Mysql', 'DBName', 'test')

    def _get_connection(self):
        try:
            return pymysql.connect(host=self._mysql_ip,
                                   port=self._mysql_port,
                                   user=self._mysql_user,
                                   password=self._mysql_pwd,
                                   database=self._mysql_db,
                                   charset='utf8')
        except Exception as e:
            logging.error(f"Failed to connect to MySQL: {e}")
            return None

    def execute(self, sql, params=None):
        with self._get_connection() as conn:
            with conn.cursor() as cursor:
                try:
                    if params:
                        cursor.execute(sql, params)
                    else:
                        cursor.execute(sql)
                    conn.commit()
                    return cursor.rowcount
                except Exception as e:
                    conn.rollback()
                    logging.error(f"Error executing SQL: {sql}, {params}. Error: {e}")
                    traceback.print_exc()
                    return -10

    def query(self, sql, params=None):
        with self._get_connection() as conn:
            with conn.cursor() as cursor:
                try:
                    if params:
                        cursor.execute(sql, params)
                    else:
                        cursor.execute(sql)
                    columns = [col[0] for col in cursor.description]
                    results = cursor.fetchall()
                    return [dict(zip(columns, row)) for row in results]
                except Exception as e:
                    logging.error(f"Error querying SQL: {sql}, {params}. Error: {e}")
                    traceback.print_exc()
                    return []

    def get_station_by_id(self, station_id):
        sql = "SELECT station_no AS ID, name AS NAME, description AS INFO, graph_id AS GRAPH_ID FROM fdp_power_station WHERE is_deleted=0 AND station_no=%s"
        return self.query(sql, [station_id])

    def get_station(self):
        sql = "SELECT station_no AS ID, name AS NAME, description AS INFO, graph_id AS GRAPH_ID FROM fdp_power_station WHERE is_deleted=0"
        return self.query(sql)

if __name__ == '__main__':
    mysql_client = MyMysql()
    result = mysql_client.get_station()
    print(result)

Key Points Analysis

Singleton implementation : Uses a @singleton decorator to simplify the pattern and keep the class definition clean.

Exception handling & logging : Wraps SQL execution in try/except blocks, logs errors, and prints stack traces for rapid debugging.

Connection management : Employs context managers (with statements) to automatically open and close connections, preventing resource leaks.

SQL execution : Provides separate execute (for INSERT/UPDATE/DELETE) and query (for SELECT) methods, supporting parameterized queries for security.

Usage example : Demonstrates how to retrieve all stations or a specific station by ID using the helper methods.

Conclusion

By following the design and code presented, you can create a robust, maintainable MySQL utility class that streamlines database interactions, enhances code readability, and improves overall application stability.