一、概述 

这个 MysqlHandler 类是一个精心设计的 MySQL 数据库操作辅助工具。它的核心目标是简化和安全化与 pymysql 库的交互。

它最大的特点是实现了 Python 的上下文管理器协议（即 __enter__ 和 __exit__ 方法），这使得数据库的连接、事务处理（提交/回滚）和资源关闭（连接/游标）能够自动、安全地完成。这极大地避免了常见的编程错误，如忘记关闭连接导致的资源泄露。

二、核心设计：上下文管理器

这是理解这个类的关键。它被设计为与 with 语句一起使用。

1. __init__(self, ...) - 初始化阶段

   • 功能: 构造函数。当你创建 MysqlHandler 的实例时，它会被调用。

   • 作用: 它不建立实际的数据库连接。它只是接收并保存所有连接所需的配置信息（如主机、用户、密码、数据库名等）到一个名为 self.config 的字典中。同时，它将 self.conn (连接) 和 self.cursor (游标) 两个属性初始化为 None，表示此时还没有连接。

2. __enter__(self) - 进入阶段

   • 功能: 当代码执行进入 with MysqlHandler(...) as db: 语句块时，这个方法会自动被调用。

   • 作用:

     • 使用 __init__ 中保存的 self.config 配置，调用 pymysql.connect() 来建立一个真实的数据库连接，并将其赋值给 self.conn。

     • 基于这个连接，创建一个游标对象，并赋值给 self.cursor。

     • return self 使得在 with 块内部，你可以使用 db 这个变量来调用类的其他方法（如 db.execute_query()）。

     • 如果连接失败，它会捕获底层的 pymysql.MySQLError，并抛出一个更具体的、包含数据库名的 ConnectionError，使错误信息更清晰。

3. __exit__(self, exc_type, ...) - 退出阶段

   • 功能: 无论 with 块是正常结束还是因异常中断，当代码执行离开 with 块时，这个方法都必定会被自动调用。

   • 作用: 这是一个自动化的清理和事务管理中心。

     • 事务管理: 它会检查 exc_type 参数。如果为 None（表示 with 块中没有发生任何异常），它会自动执行 self.conn.commit() 来提交事务。如果 exc_type 不为 None（表示发生了异常），它会执行 self.conn.rollback() 来回滚事务，防止脏数据写入。

     • 资源清理: 在事务处理之后，它会依次关闭游标 (self.cursor.close()) 和数据库连接 (self.conn.close())，确保资源被完全释放。

三、方法详解

这个类提供了一系列方法来执行常见的数据库操作

安全与易用性特性

• 'cursorclass': pymysql.cursors.DictCursor: 这是一个非常重要的配置。它让所有查询返回的结果都是字典列表 ([{'col1': val1}, ...])，而不是默认的元组列表 ([(val1, ...), ...])。这使得代码可以通过列名（如 row['id']）访问数据，极大地提高了代码的可读性和可维护性。

• _check_cursor(self): 这是一个内部的安全检查方法。所有执行数据库操作的方法在开始时都会调用它，以确保 self.cursor 不是 None。这强制开发者必须在 with 块内使用这些方法，从而避免了在没有有效连接的情况下进行操作。

通用执行方法

• execute_query(self, sql, params): 用于执行 SELECT 类型的查询，并返回所有结果行（一个字典列表）。

• execute_no_query(self, sql, params): 用于执行 INSERT, UPDATE, DELETE 等不返回数据行的“非查询”操作。它返回一个整数，表示受影响的行数。

• fetch_one(self, sql, params): 用于执行 SELECT 查询，但只返回第一条结果（一个字典），如果查询没有结果，则返回 None。

高级辅助方法

这些方法在通用方法的基础上提供了更便捷的接口，使用者无需手动编写 SQL 语句。

• insert(self, table, data): 简化了插入操作。你只需要提供表名和一个包含列名和值的字典，它会自动生成并执行 INSERT 语句。

• update(self, table, data, condition, params): 简化了更新操作。你提供表名、要更新的数据字典、一个 WHERE 条件字符串以及对应的参数。

• delete(self, table, condition, params): 简化了删除操作。你提供表名、一个 WHERE 条件字符串和对应的参数。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# @Time    : 2025/9/19 16:00
# @Author  : yuye_1990
# @Project : PyCodes
# @File    : mysql_helper.py
# @Software: PyCharm
# @Desc    : pymysql+mysql

import pymysql
# from pymysql import cursors
# PS:from pymysql import cursors 如果不显式导入，pymysql.cursors 可能无法被PyCharm等工具正确识别
from typing import Optional, Dict, Tuple, List, Any


class MysqlHandler():
    """
    一个基于上下文管理器的 MySQL 数据库连接助手类。

    此类提供了简单且安全的方式来与 MySQL 数据库交互。
    它使用上下文管理器自动处理连接和游标管理，确保资源始终正确释放。
    """

    def __init__(self, host, user, password, database, port=3306, charset='utf8mb4'):
        self.config = {
            'host': host,
            'user': user,
            'password': password,
            'database': database,
            'port': port,
            'charset': charset,
            'cursorclass': pymysql.cursors.DictCursor

            #   'cursorclass': pymysql.cursors.DictCursor
            #   这个 DictCursor 的作用是，当你获取查询结果时，它会自动将每一行数据打包成一个字典。
            #   改变数据库查询结果的返回格式，从默认的**元组（Tuple）格式变为更易于使用的字典（Dictionary）**格式
            #   cursor.execute("SELECT id, name, email FROM users WHERE id = 1") 查询结果差别:
            #   (1, 'Alice', 'alice@example.com')  vs {'id': 1, 'name': 'Alice', 'email': 'alice@example.com'}

        }
        self.conn: Optional[pymysql.connections.Connection] = None
        self.cursor: Optional[pymysql.cursors.Cursor] = None

    #     : Optional[pymysql.connections.Connection]  和 : Optional[pymysql.cursors.Cursor]
    #     这两行代码是 Python 的类型提示 (Type Hinting)，
    #     作用是为类属性（self.conn 和 self.cursor）提供明确的类型信息，以便于开发者和开发工具（如 PyCharm）理解代码

    def __enter__(self):
        """
        进入 'with' 块时建立数据库连接。
        """
        try:
            self.conn = pymysql.connect(**self.config)
            self.cursor = self.conn.cursor()
            return self
        except pymysql.MySQLError as e:
            db_name = self.config['database']
            raise ConnectionError(f'{db_name} 数据库连接错误。错误为： {e}') from e

    def __exit__(self, exc_type, exc_val, exc_tb):
        """
        退出 'with' 块时关闭游标和连接。
        成功时提交事务，失败时回滚。
        """
        if exc_type is None:
            if self.conn:
                self.conn.commit()
        else:
            if self.conn:
                self.conn.rollback()
        if self.cursor:
            self.cursor.close()
        if self.conn:
            self.conn.close()

    def _check_cursor(self):
        """内部辅助方法，确保游标可用。"""
        if not self.cursor:
            raise ConnectionError("游标不可用。数据库操作必须在 'with' 代码块内执行。")

    # 作用: 这是对错误的根本原因的解释，并且直接给出了修正方案。
    # 只有在进入 with 代码块时，__enter__ 方法才会被调用，此时 self.conn 和 self.cursor 才会被创建和赋值。
    # 当代码块结束时，__exit__ 方法会被调用，游标和连接会被关闭（在更完善的设计中，它们可能还会被重置为 None）。
    # 因此，游标只在 with 代码块的内部才是可用的、“存活”的状态。任何试图在 with 块之外调用 execute_query、insert 等数据操作方法的行为，都是对这个类的错误使用。
    # 这句提示就是为了捕获这种错误用法，并友好地“教育”使用者：“嘿，你用错地方了！你应该把这个数据库操作放到一个 with 语句里面去。

    def execute_query(self, sql: str, params: Optional[Tuple] = None) -> List[Dict[str, Any]]:
        """执行 SQL 查询 (SELECT) 并返回所有结果。"""
        self._check_cursor()
        self.cursor.execute(sql, params or ())
        return self.cursor.fetchall()

    # params or () 给 execute 方法提供一个安全的、永不为 None 的参数默认值，从而避免程序因 TypeError 而崩溃。
    # cursor.execute 方法的“契约”：它接收两个参数：execute(sql_string, parameters)
    # 第一个参数 sql_string 是要执行的 SQL 语句。第二个参数 parameters 是一个可选的、用来替换 SQL 语句中占位符（如 %s）的序列（通常是元组或列表）。
    # 数据库驱动（如 pymysql）的 execute 方法可以接受一个空元组 () （表示没有参数），但不能接受 None 作为第二个参数。如果尝试传入 None，它会抛出一个 TypeError。

    # PS:如果 sql 字符串是一个完整的、不包含任何参数占位符（如 %s）的静态语句，那么 cursor.execute(sql) 是完全正确的
    # PS:如果 sql 字符串包含了参数占位符 (%s)，那么 cursor.execute(sql) 会出错,需要提供第二个参数

    def execute_no_query(self, sql: str, params: Optional[Tuple] = None) -> int:
        """执行数据修改类 SQL 语句 (INSERT, UPDATE, DELETE)。"""
        self._check_cursor()
        return self.cursor.execute(sql, params or ())

    def fetch_one(self, sql:str, params:Optional[Tuple] = None) -> Optional[Dict[str, Any]]:
        """执行 SQL 查询并获取单条记录。"""
        self._check_cursor()
        self.cursor.execute(sql, params or ())
        return self.cursor.fetchone()

    def insert(self, table: str, data: Dict):
        self._check_cursor()
        if not data:
            raise ValueError("用于插入的数据字典 (data) 不能为空。")
        # columns = ','.join(data.keys())
        columns = ', '.join([f"`{key}`" for key in data.keys()])
        placeholders = ','.join(['%s'] * len(data))
        sql = f"INSERT INTO `{table}`({columns}) VALUES ({placeholders}) "
        return self.execute_no_query(sql, tuple(data.values()))

    def update(self, table: str, data: Dict, condition: str, params: Optional[Tuple] = None) -> int:
        set_clause = ','.join([f"`{key}`=%s" for key in data.keys()])
        sql = f"UPDATE `{table}` SET {set_clause} WHERE {condition}"
        all_params = tuple(data.values()) + (params or ())
        return self.execute_no_query(sql, all_params)

    def delete(self, table: str, condition: str, params: Optional[Tuple] = None) -> int:
        self._check_cursor()
        if not condition or not params:
            raise ValueError("为安全起见，删除操作需要明确的 condition 和 params。")
        sql = f'DELETE FROM `{table}` WHERE {condition}'
        return self.execute_no_query(sql, params or ())
    # 提示：update 和 delete 方法的 condition 参数仍然是直接拼接的，存在相同的潜在安全风险点，需调用者保证 condition 字符串的来源安全