Python使用pymysql封装数据库访问类

Python中使用pymysql库封装数据库访问类是一种常见的做法，它可以简化数据库操作，同时确保资源的正确释放。以下是一个基本的数据库连接类封装示例：

import pymysql
from pymysql.cursors import DictCursor
from warnings import filterwarnings

# 忽略MySQL警告
filterwarnings("ignore", category=pymysql.Warning)

class DatabaseManager:
    def __init__(self, host, user, password, db_name, charset='utf8mb4'):
        """
        初始化数据库连接
        """
        self.conn = None
        self.cursor = None

        # 连接数据库
        self.conn = pymysql.connect(
            host=host,
            user=user,
            password=password,
            db=db_name,
            charset=charset,
            cursorclass=DictCursor
        )
        # 获取操作游标
        self.cursor = self.conn.cursor()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        # 如果有异常发生，回滚事务
        if exc_val:
            self.conn.rollback()
        # 关闭游标
        if self.cursor:
            self.cursor.close()
        # 关闭连接
        if self.conn:
            self.conn.close()

    def execute_query(self, query, params=None, fetch_all=True):
        """
        执行查询并获取结果
        """
        try:
            self.cursor.execute(query, params or ())
            if fetch_all:
                results = self.cursor.fetchall()
            else:
                results = self.cursor.fetchone()
            return results
        except Exception as e:
            print(f"数据库查询执行异常：{e}")
            raise

    def execute_statement(self, statement, params=None):
        """
        执行非查询语句（如INSERT, UPDATE, DELETE等）
        """
        try:
            affected_rows = self.cursor.execute(statement, params or ())
            # 提交事务
            self.conn.commit()
            return affected_rows
        except Exception as e:
            print(f"数据库操作执行异常：{e}")
            # 如果操作失败，回滚事务
            self.conn.rollback()
            raise
    def update(self, update_query, params=None):
        """
        执行UPDATE语句
        """
        return self.execute_statement(update_query, params)

    def delete(self, delete_query, params=None):
        """
        执行DELETE语句
        """
        return self.execute_statement(delete_query, params)

    def create_table(self, create_table_query):
        """
        创建表
        """
        try:
            self.cursor.execute(create_table_query)
            self.conn.commit()
            print("表创建成功")
        except pymysql.MySQLError as e:
            print(f"创建表时发生错误: {e}")
            self.conn.rollback()
# 使用示例
with DatabaseManager('127.0.0.1', 'username', 'password', 'database_name') as db:
    # 查询数据
    result = db.execute_query("SELECT * FROM your_table")
    print(result)

    # 插入数据
    db.execute_statement("INSERT INTO your_table (column1, column2) VALUES (%s, %s)", ('value1', 'value2'))

    # 更新数据
    updated_rows = db.update("UPDATE your_table SET column1 = %s WHERE id = %s", ('new_value', 1))
    print(f"更新了{updated_rows}行数据")

    # 删除数据
    deleted_rows = db.delete("DELETE FROM your_table WHERE id = %s", (2,))
    print(f"删除了{deleted_rows}行数据")

在这个示例中：

• 类DatabaseManager负责初始化数据库连接以及获取游标。
• 它实现了上下文管理协议（即__enter__和__exit__方法），使得可以用with语句来确保即使在发生异常的情况下也能正确关闭连接和游标。
• execute_query方法用于执行SQL查询语句并获取结果。
• execute_statement方法用于执行那些更改数据库状态的SQL语句（如INSERT, UPDATE, DELETE），并且会在操作成功后提交事务，如果出现异常，则会回滚事务。