当前位置：首页 > news >正文

使用 Python 操作 MySQL 数据库的实用工具类：MySQLHandler

news 2025/7/7 3:52:32

操作数据库是非常常见的需求，使用 Python 和 pymysql 库封装一个通用的 MySQL 数据库操作工具类，并通过示例演示如何使用这个工具类高效地管理数据库。

工具类的核心代码解析

MySQLHandler 类简介

MySQLHandler 是一个 Python 类，用于简化对 MySQL 数据库的常见操作，如连接、查询、插入、更新和删除等。以下是该类的主要功能：

连接数据库

使用 pymysql.connect 方法建立与数据库的连接。

执行查询

提供 execute_query 方法执行 SELECT 查询，返回查询结果。

数据插入

使用 insert_data 方法将数据插入指定的表。

数据更新与删除

提供 update_data 和 delete_data 方法分别用于更新和删除数据。

代码详解

以下是 mysql_utils.py 的完整代码：

import pymysqlclass MySQLHandler:def __init__(self, host, user, password, database, port=3306):self.host = hostself.user = userself.password = passwordself.database = databaseself.port = portself.connection = Nonedef connect(self):"""建立数据库连接"""try:self.connection = pymysql.connect(host=self.host,user=self.user,password=self.password,database=self.database,port=self.port,charset='utf8mb4')except Exception as e:print(f"连接数据库失败: {e}")raisedef close(self):"""关闭数据库连接"""if self.connection:self.connection.close()def execute_query(self, query, params=None):"""执行查询语句"""try:with self.connection.cursor() as cursor:cursor.execute(query, params)result = cursor.fetchall()return resultexcept Exception as e:print(f"查询失败: {e}")raisedef execute_update(self, query, params=None):"""执行更新语句"""try:with self.connection.cursor() as cursor:rows_affected = cursor.execute(query, params)self.connection.commit()return rows_affectedexcept Exception as e:print(f"更新失败: {e}")self.connection.rollback()raisedef insert_data(self, table, data):"""插入数据"""try:keys = ', '.join(data.keys())values = ', '.join(['%s'] * len(data))query = f"INSERT INTO {table} ({keys}) VALUES ({values})"self.execute_update(query, list(data.values()))except Exception as e:print(f"插入数据失败: {e}")raisedef delete_data(self, table, condition, params):"""删除数据"""try:query = f"DELETE FROM {table} WHERE {condition}"self.execute_update(query, params)except Exception as e:print(f"删除数据失败: {e}")raisedef update_data(self, table, updates, condition, params):"""更新数据"""try:update_str = ', '.join([f"{key}=%s" for key in updates.keys()])query = f"UPDATE {table} SET {update_str} WHERE {condition}"self.execute_update(query, list(updates.values()) + params)except Exception as e:print(f"更新数据失败: {e}")raise

使用示例

main.py 文件提供了对 MySQLHandler 的使用示例，包含常见的数据库操作流程。

from mysql_utils import MySQLHandlerdef main():# 创建 MySQLHandler 实例db = MySQLHandler(host="localhost", user="root", password="root", database="testdb")try:# 连接数据库db.connect()# 执行查询示例result = db.execute_query("SELECT * FROM users")print("查询结果:", result)# 插入数据示例db.insert_data("users", {"name": "Alice", "age": 25})print("数据插入成功")# 更新数据示例db.update_data("users", {"age": 26}, "name=%s", ["Alice"])print("数据更新成功")# 删除数据示例db.delete_data("users", "name=%s", ["Alice"])print("数据删除成功")except Exception as e:print(f"发生错误: {e}")finally:# 关闭数据库连接db.close()if __name__ == "__main__":main()