Python MySQL Wrapper: Encapsulating Database Operations

Introduction Encapsulating MySQL operations in a dedicated wrapper class makes your code more modular, reusable, and easier to maintain. By hiding the low‑level database interaction behind a clean interface, you can focus on business logic while the wrapper handles connections, queries, and error handling.

Complete 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 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 operation."""
        return self.execute_query(query, values, commit=True)

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

    def delete_data(self, query, values):
        """Execute a DELETE operation."""
        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 of the Class

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

__init__ stores connection parameters.

connect establishes the connection.

close shuts it down.

execute_query runs any SQL statement, handling SELECT results or committing changes for INSERT/UPDATE/DELETE.

Convenience methods query_data , insert_data , update_data , and delete_data wrap common CRUD operations.

Main Function

Instantiates MySQLWrapper and connects to the database.

Demonstrates SELECT, INSERT, UPDATE, and DELETE operations using the wrapper.

Closes the connection at the end.

Running the Code

Install the required library: pip install mysql-connector-python

Save the script to a file (e.g., mysql_wrapper.py ) and run it with python mysql_wrapper.py .

Conclusion

By following these steps, you have created a reusable MySQL wrapper that simplifies connection management and CRUD operations, improving code readability, maintainability, and extensibility for future enhancements such as transaction handling or advanced error processing.