Performing MySQL CRUD Operations in Python Using INI Files and Logging

When developing database‑driven applications, performing CRUD (Create, Read, Update, Delete) operations is essential. Python offers several ways to interact with MySQL, and this guide shows how to keep SQL statements in .ini files, read them with configparser , and execute them using the mysql-connector-python library while logging every operation.

Preparation

First, install the required packages:

pip install mysql-connector-python

Create two configuration files:

db_config.ini – database connection information

[mysql]
host=localhost
user=root
passwd=password
database=testdb

sql_queries.ini – SQL statements

[Queries]
create_employee = INSERT INTO employees (name, position, office, salary) VALUES (%s, %s, %s, %s)
read_employees = SELECT * FROM employees
update_employee_salary = UPDATE employees SET salary = %s WHERE id = %s
delete_employee = DELETE FROM employees WHERE id = %s

Encapsulating database operations

The following database_manager.py script manages the connection, reads queries from the .ini file, executes CRUD actions, and logs each step.

import configparser
import mysql.connector
from mysql.connector import Error
import logging

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

class DatabaseManager:
    def __init__(self, config_file='db_config.ini', sql_file='sql_queries.ini'):
        self.config = configparser.ConfigParser()
        self.sql_queries = configparser.ConfigParser()
        self.config.read(config_file)
        self.sql_queries.read(sql_file)
        self.connection = None

    def create_connection(self):
        """Create a database connection"""
        try:
            self.connection = mysql.connector.connect(
                host=self.config.get('mysql', 'host'),
                user=self.config.get('mysql', 'user'),
                passwd=self.config.get('mysql', 'passwd'),
                database=self.config.get('mysql', 'database')
            )
            print("连接到 MySQL 数据库成功")
            logging.info("连接到 MySQL 数据库成功")
        except Error as e:
            print(f"发生错误 '{e}'")
            logging.error(f"发生错误 '{e}'")

    def close_connection(self):
        """Close the database connection"""
        if self.connection:
            self.connection.close()
            print("数据库连接已关闭")
            logging.info("数据库连接已关闭")

    def execute_query(self, query_name, data=None):
        """Execute an SQL query and log it"""
        cursor = self.connection.cursor()
        try:
            query = self.sql_queries.get('Queries', query_name)
            logging.info(f"执行 SQL: {query} | 数据: {data}")
            if data:
                cursor.execute(query, data)
            else:
                cursor.execute(query)
            self.connection.commit()
            print("查询执行成功")
            logging.info("查询执行成功")
        except Error as e:
            print(f"发生错误 '{e}'")
            logging.error(f"发生错误 '{e}'")
        finally:
            cursor.close()

    def fetch_data(self, query_name):
        """Fetch data and log it"""
        cursor = self.connection.cursor(dictionary=True)
        try:
            query = self.sql_queries.get('Queries', query_name)
            logging.info(f"执行 SQL: {query}")
            cursor.execute(query)
            result = cursor.fetchall()
            return result
        except Error as e:
            print(f"发生错误 '{e}'")
            logging.error(f"发生错误 '{e}'")
        finally:
            cursor.close()

# Example usage
if __name__ == "__main__":
    db_manager = DatabaseManager()
    db_manager.create_connection()
    # Insert a new employee
    employee_data = ('John Doe', 'Developer', 'London', 123000)
    db_manager.execute_query('create_employee', employee_data)
    # Read all employees
    employees = db_manager.fetch_data('read_employees')
    for employee in employees:
        print(employee)
    # Update employee salary
    new_salary = (150000, 1)
    db_manager.execute_query('update_employee_salary', new_salary)
    # Delete an employee
    employee_id = (1,)
    db_manager.execute_query('delete_employee', employee_id)
    db_manager.close_connection()

Log configuration explanation

The log file database_operations.log records messages with level INFO and above, using the format %(asctime)s - %(levelname)s - %(message)s , which includes a timestamp, the severity level, and the log message.

Conclusion

By following these steps, you can easily perform MySQL CRUD operations with Python, keep all SQL actions logged for transparency, and improve code maintainability and security.