### 封装 MySQL 操作的方法 为了简化 Python 对 MySQL 数据库的操作，可以创建一个封装类来处理常见的 CRUD（增删改查）操作。这不仅提高了代码的可读性和维护性，还减少了重复代码的数量。 #### 创建 `MySQLHelper` 类 下面是一个基于 `pymysql` 库构建的基础版 `MySQLHelper` 类： ```python import pymysql class MySQLHelper: def __init__(self, host='localhost', port=3306, user='', password='', db=''): """ 初始化数据库连接参数 :param host: 主机地址，默认为本地主机 :param port: 端口号，默认为 3306 :param user: 用户名 :param password: 密码 :param db: 要使用的数据库名称 """ self.host = host self.port = port self.user = user self.password = password self.db = db try: self.connection = pymysql.connect( host=self.host, port=self.port, user=self.user, passwd=self.password, database=self.db ) except Exception as e: print(f"Failed to connect to the database {e}") def execute_query(self, sql, params=None): """ 执行查询并返回结果集 :param sql: SQL 查询语句 :param params: 参数列表或字典形式的数据 :return: 结果集 """ with self.connection.cursor() as cursor: cursor.execute(sql, params) result = cursor.fetchall() return result def insert_data(self, table_name, data_tuple): """ 插入单条记录到指定表中 :param table_name: 表名 :param data_tuple: 待插入的数据元组 :return: 影响行数 """ placeholders = ', '.join(['%s'] * len(data_tuple)) columns = ', '.join([f"`{col}`" for col in data_tuple._fields]) query = f"INSERT INTO `{table_name}` ({columns}) VALUES ({placeholders})" affected_rows = None with self.connection.cursor() as cursor: affected_rows = cursor.execute(query, tuple(data_tuple)) self.connection.commit() return affected_rows def update_by_id(self, table_name, id_field, record_id, updates_dict): """ 更新特定 ID 的记录 :param table_name: 表名 :param id_field: 主键字段名 :param record_id: 记录ID值 :param updates_dict: 需要更新的内容，key-value对的形式给出 :return: 影响行数 """ set_clause = ", ".join([f"{k}=%({k})s" for k in updates_dict.keys()]) where_clause = f"{id_field}=%(record_id)s" query = f"UPDATE `{table_name}` SET {set_clause} WHERE {where_clause}" affected_rows = None with self.connection.cursor() as cursor: combined_params = {'record_id': record_id} combined_params.update(updates_dict) affected_rows = cursor.execute(query, combined_params) self.connection.commit() return affected_rows def delete_by_id(self, table_name, id_field, record_id): """ 删除具有给定 ID 的记录 :param table_name: 表名 :param id_field: 主键字段名 :param record_id: 记录ID值 :return: 影响行数 """ query = f"DELETE FROM `{table_name}` WHERE `{id_field}`=%(record_id)s;" affected_rows = None with self.connection.cursor() as cursor: affected_rows = cursor.execute(query, {"record_id": record_id}) self.connection.commit() return affected_rows def fetch_one_by_id(self, table_name, id_field, record_id): """ 获取某张表里某个具体 ID 的一条数据 :param table_name: 表名 :param id_field: 主键字段名 :param record_id: 记录ID值 :return: 单一的结果对象 """ query = f"SELECT * FROM `{table_name}` WHERE `{id_field}`=%(record_id)s LIMIT 1;" with self.connection.cursor(pymysql.cursors.DictCursor) as cursor: cursor.execute(query, {"record_id": record_id}) row = cursor.fetchone() return row def close_connection(self): """关闭数据库连接""" if hasattr(self, 'connection') and self.connection.open: self.connection.close() ``` 此版本提供了基本的功能支持，包括但不限于插入、删除、更新以及按条件检索数据等功能[^2]。对于更复杂的业务逻辑，则可以根据实际需求进一步扩展此类的功能。