环境准备

pip install pymysql

具体代码

import pymysql
from pymysql.cursors import DictCursor

class MySQLHandler:
    def __init__(self, host, user, password, database, port=3306, charset='utf8mb4'):
        self.host = host
        self.user = user
        self.password = password
        self.database = database
        self.port = port
        self.charset = charset
        self.connection = None

    def connect(self):
        """建立数据库连接"""
        try:
            self.connection = pymysql.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database,
                port=self.port,
                charset=self.charset,
                cursorclass=DictCursor  # 返回字典类型结果
            )
        except pymysql.MySQLError as e:
            print(f"连接数据库失败: {e}")
            raise

    def close(self):
        """关闭连接"""
        if self.connection:
            self.connection.close()

    def execute_query(self, sql, params=None):
        """执行查询语句（返回多条数据）"""
        try:
            with self.connection.cursor() as cursor:
                cursor.execute(sql, params)
                return cursor.fetchall()
        except pymysql.MySQLError as e:
            print(f"查询失败: {e}")
            self.connection.rollback()
            raise

    def execute_update(self, sql, params=None):
        """执行更新语句（插入/更新/删除）"""
        try:
            with self.connection.cursor() as cursor:
                rows_affected = cursor.execute(sql, params)
                self.connection.commit()
                return rows_affected
        except pymysql.MySQLError as e:
            print(f"更新失败: {e}")
            self.connection.rollback()
            raise

    def get_one(self, sql, params=None):
        """查询单条数据"""
        try:
            with self.connection.cursor() as cursor:
                cursor.execute(sql, params)
                return cursor.fetchone()
        except pymysql.MySQLError as e:
            print(f"查询单条数据失败: {e}")
            self.connection.rollback()
            raise