pip install mysql-connector-python

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

[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

import configparser
import mysql.connector
from mysql.connector import Error
import logging
# 設(shè)置日志配置
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):
        """創(chuàng)建數(shù)據(jù)庫連接"""
        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 數(shù)據(jù)庫成功")
            logging.info("連接到 MySQL 數(shù)據(jù)庫成功")
        except Error as e:
            print(f"發(fā)生錯誤 '{e}'")
            logging.error(f"發(fā)生錯誤 '{e}'")
    def close_connection(self):
        """關(guān)閉數(shù)據(jù)庫連接"""
        if self.connection:
            self.connection.close()
            print("數(shù)據(jù)庫連接已關(guān)閉")
            logging.info("數(shù)據(jù)庫連接已關(guān)閉")
    def execute_query(self, query_name, data=None):
        """執(zhí)行 SQL 查詢并記錄日志"""
        cursor = self.connection.cursor()
        try:
            query = self.sql_queries.get('Queries', query_name)
            logging.info(f"執(zhí)行 SQL: {query} | 數(shù)據(jù): {data}")
            if data:
                cursor.execute(query, data)
            else:
                cursor.execute(query)
            self.connection.commit()
            print("查詢執(zhí)行成功")
            logging.info("查詢執(zhí)行成功")
        except Error as e:
            print(f"發(fā)生錯誤 '{e}'")
            logging.error(f"發(fā)生錯誤 '{e}'")
        finally:
            cursor.close()
    def fetch_data(self, query_name):
        """獲取數(shù)據(jù)并記錄日志"""
        cursor = self.connection.cursor(dictinotallow=True)
        try:
            query = self.sql_queries.get('Queries', query_name)
            logging.info(f"執(zhí)行 SQL: {query}")
            cursor.execute(query)
            result = cursor.fetchall()
            return result
        except Error as e:
            print(f"發(fā)生錯誤 '{e}'")
            logging.error(f"發(fā)生錯誤 '{e}'")
        finally:
            cursor.close()
# 示例用法：
if __name__ == "__main__":
    db_manager = DatabaseManager()
    db_manager.create_connection()
    # 插入新員工記錄
    employee_data = ('John Doe', 'Developer', 'London', 123000)
    db_manager.execute_query('create_employee', employee_data)
    # 讀取所有員工信息
    employees = db_manager.fetch_data('read_employees')
    for employee in employees:
        print(employee)
    # 更新員工薪資
    new_salary = (150000, 1)
    db_manager.execute_query('update_employee_salary', new_salary)
    # 刪除員工記錄
    employee_id = (1,)
    db_manager.execute_query('delete_employee', employee_id)
    db_manager.close_connection()