Python MySQL Wrapper: Encapsulating Database Operations

Preface

Encapsulating MySQL operations can make your code more modular, reusable, and easier to maintain. By wrapping the database interaction details, you expose a clean interface for other parts of your application.

Full Code

import mysql.connector
from mysql.connector import Error

class MySQLWrapper:
    def __init__(self, host, user, password, database):
        """Initialize database connection parameters."""
        self.host = host
        self.user = user
        self.password = password
        self.database = database
        self.connection = None

    def connect(self):
        """Connect to the MySQL database."""
        try:
            self.connection = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database
            )
            if self.connection.is_connected():
                db_info = self.connection.get_server_info()
                print(f"Connected to MySQL Server version {db_info}")
        except Error as e:
            print(f"Error while connecting to MySQL: {e}")
            self.connection = None

    def close(self):
        """Close the database connection."""
        if self.connection and self.connection.is_connected():
            self.connection.close()
            print("MySQL connection is closed.")

    def execute_query(self, query, values=None, commit=False):
        """Execute an SQL statement and optionally return results or commit changes."""
        if not self.connection or not self.connection.is_connected():
            print("Database connection is not established.")
            return None
        cursor = self.connection.cursor()
        try:
            if values:
                cursor.execute(query, values)
            else:
                cursor.execute(query)
            if commit:
                self.connection.commit()
            if query.strip().upper().startswith('SELECT'):
                return cursor.fetchall()
            else:
                return cursor.rowcount
        except Error as e:
            print(f"Error executing SQL: {e}")
            return None
        finally:
            cursor.close()

    def query_data(self, query, values=None):
        """Execute a SELECT query."""
        return self.execute_query(query, values, commit=False)

    def insert_data(self, query, values):
        """Execute an INSERT query."""
        return self.execute_query(query, values, commit=True)

    def update_data(self, query, values):
        """Execute an UPDATE query."""
        return self.execute_query(query, values, commit=True)

    def delete_data(self, query, values):
        """Execute a DELETE query."""
        return self.execute_query(query, values, commit=True)

def main():
    # Create MySQLWrapper instance and connect to the database
    db = MySQLWrapper(host='localhost', user='root', password='password', database='testdb')
    db.connect()
    if db.connection is not None:
        # Query data
        query = "SELECT * FROM users WHERE age > %s"
        results = db.query_data(query, (18,))
        print("Query Results:")
        for row in results:
            print(row)
        # Insert data
        insert_query = "INSERT INTO users (name, age) VALUES (%s, %s)"
        insert_values = ("John Doe", 25)
        rows_affected = db.insert_data(insert_query, insert_values)
        print(f"Insert affected {rows_affected} rows.")
        # Update data
        update_query = "UPDATE users SET age = %s WHERE name = %s"
        update_values = (26, "John Doe")
        rows_affected = db.update_data(update_query, update_values)
        print(f"Update affected {rows_affected} rows.")
        # Delete data
        delete_query = "DELETE FROM users WHERE name = %s"
        delete_values = ("John Doe",)
        rows_affected = db.delete_data(delete_query, delete_values)
        print(f"Delete affected {rows_affected} rows.")
        # Close the connection
        db.close()

if __name__ == '__main__':
    main()

Explanation

The MySQLWrapper class encapsulates all methods needed to interact with a MySQL database.

__init__ initializes connection parameters.

connect establishes the database connection.

close terminates the connection.

execute_query runs any SQL statement and optionally returns results or commits changes.

query_data , insert_data , update_data , and delete_data provide convenient wrappers for SELECT, INSERT, UPDATE, and DELETE operations respectively.

Main Function

Creates an instance of MySQLWrapper , connects to the database, performs a SELECT query, inserts a new record, updates it, deletes it, and finally closes the connection.

Summary

By following these steps, we have encapsulated common MySQL operations—connection handling, SQL execution, querying, inserting, updating, and deleting—into a reusable Python class, improving code readability and maintainability. The wrapper can be further extended with transaction management, advanced error handling, and additional utility methods as needed.