## docker run --name my-mysql -e MYSQL_ROOT_PASSWORD=happy23 -d -p 3307:3306 mysql:8.0.41
    docker run --name my-mysql -e MYSQL_ROOT_PASSWORD=happy23 -d -p 3307:3306 mysql:latest
# 四，拉取mysql 镜像地址
    bash
    docker pull mysql:latest
    或直接运行：
    
    bash
    docker run --name my-mysql -e MYSQL_ROOT_PASSWORD=happy23 -d -p 3307:3306 mysql:latest

### 下载完成后，检查容器状态
    bash
### 查看容器是否在运行
    docker ps

### 应该能看到名为 my-mysql 的容器，状态为 Up
    连接测试
    容器运行后，用以下方式测试连接：

### 方式一：通过命令行连接

    bash
    docker exec -it my-mysql mysql -uroot -phappy23
### 方式二：通过 Windows 客户端连接

    地址：localhost 或 127.0.0.1
    
    端口：3307
    
    用户名：root
    
    密码：happy23


# =================================================

### 在 MySQL 里，root 是一个特殊的超级管理员账号，官方不推荐也不支持直接将其“重命名”。不过，我们可以通过另一种方式实现同样的目的：创建一个拥有全部权限、可以替代 root 的新用户，然后停用 root 用户。

### 下面是在你的 Docker MySQL 容器中实现这个目标的具体步骤。

# 操作流程
#### 进入容器：在终端执行以下命令，进入 MySQL 容器的命令行界面，my-mysql 替换成你的容器名：

bash
    docker exec -it my-mysql mysql -u root -p
    然后输入你之前设置的 happy23 密码。

#### 创建新管理用户：在 MySQL 命令行中执行以下命令，创建一个名为 newadmin，密码为 your_strong_password 的新用户，并赋予它与 root 相同的所有权限：

```sql
    -- 创建一个可以从任何主机连接的新用户
    CREATE USER 'newadmin'@'%' IDENTIFIED BY 'your_strong_password';
    
    -- 授予该用户所有数据库的所有权限
    GRANT ALL PRIVILEGES ON *.* TO 'newadmin'@'%' WITH GRANT OPTION;
    
    -- 刷新权限，使设置立即生效
    FLUSH PRIVILEGES;
    这里将 'newadmin'@'%' 中的 % 替换为 'localhost'，可以限制该用户仅能从容器内部或本地主机登录。
    
    验证并切换用户：输入 exit; 退出 MySQL 命令行，然后用新用户尝试登录，验证其功能：
```
 
#### docker exec -it my-mysql mysql -u newadmin -p
  （可选）禁用 root 用户：确认新用户可用后，回到 MySQL 命令行，执行以下命令删除 root 用户：

   ⚠️ 警告：执行此操作前，请务必确认你的新管理员账户一切正常，否则可能导致数据库无法管理。

#### sql
    -- 删除 root 用户（注意 host 可能有多个，如 'localhost', '%'）
    DROP USER 'root'@'localhost';
    DROP USER 'root'@'%';

    -- 刷新权限
    FLUSH PRIVILEGES;
## 💡 一些说明
    为什么不直接用 RENAME USER？ MySQL 的 RENAME USER 语句确实存在，但它通常不推荐用于 root 用户，因为这可能影响一些内部对象（如存储过程、视图）的定义者（DEFINER）属性，导致它们失效或产生安全问题。创建新用户是更安全和标准的做法。
    
    关于连接：创建新用户后，从 Windows 宿主机或其他客户端连接 MySQL 时，只需要将用户名从 root 换成你新创建的用户名（如 newadmin），连接地址 localhost、端口 3307 和密码保持不变。

## 📝 总结
简单来说，完成后的结果如下：

## 项目   说明
    原 root 用户   密码 happy23，默认情况下会被保留或安全移除
    新管理员用户  用户名 newadmin，密码 your_strong_password，拥有全部权限
    连接方式    将客户端或应用中的用户名从 root 替换为 newadmin 即可

import pymysql
from pymysql.cursors import DictCursor
from contextlib import contextmanager
from typing import List, Dict, Any, Optional, Union
import logging

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


class MySQLClientManager:
    """
    MySQL 数据库客户端管理类
    支持连接管理、CRUD 操作、事务处理等
    """

    def __init__(
            self,
            host: str = 'localhost',
            port: int = 3306,
            user: str = 'root',
            password: str = '',
            database: str = '',
            charset: str = 'utf8mb4',
            autocommit: bool = False,
            pool_size: int = 5,
            auto_create_db: bool = False,
            **kwargs
    ):
        """
        初始化 MySQL 连接配置

        Args:
            host: 数据库主机地址
            port: 端口号
            user: 用户名
            password: 密码
            database: 数据库名称
            charset: 字符集
            autocommit: 是否自动提交
            pool_size: 连接池大小（目前仅配置，后续可扩展）
            auto_create_db: 是否自动创建数据库（如果不存在）
            **kwargs: 其他 pymysql 参数
        """
        self.config = {
            'host': host,
            'port': port,
            'user': user,
            'password': password,
            'database': database,
            'charset': charset,
            'autocommit': autocommit,
            'cursorclass': DictCursor,
            **kwargs
        }
        self._connection = None
        self.pool_size = pool_size

        # 如果需要自动创建数据库
        if auto_create_db and database:
            self._ensure_database_exists()

        logger.info(f"MySQLClientManager 初始化完成: {host}:{port}/{database}")

    def _ensure_database_exists(self):
        """确保数据库存在，如果不存在则创建"""
        database_name = self.config['database']
        try:
            # 先不指定数据库连接
            temp_config = self.config.copy()
            temp_config.pop('database')
            temp_conn = pymysql.connect(**temp_config)
            with temp_conn.cursor() as cursor:
                cursor.execute(f"CREATE DATABASE IF NOT EXISTS {database_name}")
            temp_conn.close()
            logger.info(f"数据库 {database_name} 已创建或已存在")
        except Exception as e:
            logger.error(f"创建数据库失败: {e}")
            raise

    def _get_connection(self):
        """获取数据库连接（单例模式）"""
        if self._connection is None:
            try:
                self._connection = pymysql.connect(**self.config)
                logger.debug("数据库连接已建立")
            except pymysql.Error as e:
                logger.error(f"数据库连接失败: {e}")
                raise
        else:
            # 检查连接是否存活
            try:
                self._connection.ping(reconnect=True)
            except pymysql.Error:
                logger.warning("连接已断开，重新建立连接")
                self._connection = pymysql.connect(**self.config)
        return self._connection

    def close(self):
        """关闭数据库连接"""
        if self._connection:
            self._connection.close()
            self._connection = None
            logger.debug("数据库连接已关闭")

    @contextmanager
    def get_cursor(self):
        """
        获取游标的上下文管理器
        自动处理事务提交/回滚
        """
        conn = self._get_connection()
        cursor = conn.cursor()
        try:
            yield cursor
            if not self.config.get('autocommit', False):
                conn.commit()
        except Exception as e:
            if not self.config.get('autocommit', False):
                conn.rollback()
            logger.error(f"数据库操作失败: {e}")
            raise
        finally:
            cursor.close()

    # ==================== 查询操作 ====================

    def execute_query(self, sql: str, params: Optional[Union[tuple, dict]] = None) -> List[Dict[str, Any]]:
        """
        执行查询语句，返回所有结果

        Args:
            sql: SQL 语句
            params: 参数（tuple 或 dict）

        Returns:
            查询结果列表，每行是一个字典
        """
        with self.get_cursor() as cursor:
            cursor.execute(sql, params or ())
            result = cursor.fetchall()
            logger.debug(f"查询成功，返回 {len(result)} 行")
            return result

    def execute_query_one(self, sql: str, params: Optional[Union[tuple, dict]] = None) -> Optional[Dict[str, Any]]:
        """
        执行查询语句，返回单条结果

        Args:
            sql: SQL 语句
            params: 参数

        Returns:
            单条结果字典，如果没有结果则返回 None
        """
        with self.get_cursor() as cursor:
            cursor.execute(sql, params or ())
            result = cursor.fetchone()
            logger.debug(f"查询成功，返回 {'1 行' if result else '0 行'}")
            return result

    def execute_insert(self, table: str, data: Dict[str, Any]) -> int:
        """
        插入单条数据

        Args:
            table: 表名
            data: 要插入的数据字典 {列名: 值}

        Returns:
            插入的行 ID
        """
        columns = ', '.join(data.keys())
        placeholders = ', '.join(['%s'] * len(data))
        sql = f"INSERT INTO {table} ({columns}) VALUES ({placeholders})"

        with self.get_cursor() as cursor:
            cursor.execute(sql, tuple(data.values()))
            last_id = cursor.lastrowid
            logger.info(f"插入成功: {table}, ID={last_id}")
            return last_id

    def execute_insert_batch(self, table: str, data_list: List[Dict[str, Any]]) -> int:
        """
        批量插入数据

        Args:
            table: 表名
            data_list: 要插入的数据字典列表

        Returns:
            插入的行数
        """
        if not data_list:
            return 0

        columns = ', '.join(data_list[0].keys())
        placeholders = ', '.join(['%s'] * len(data_list[0]))
        sql = f"INSERT INTO {table} ({columns}) VALUES ({placeholders})"

        # 构建参数列表
        params_list = [tuple(d.values()) for d in data_list]

        with self.get_cursor() as cursor:
            affected = cursor.executemany(sql, params_list)
            logger.info(f"批量插入成功: {table}, 共 {affected} 行")
            return affected

    def execute_update(self, table: str, data: Dict[str, Any], where: str, where_params: Optional[tuple] = None) -> int:
        """
        更新数据

        Args:
            table: 表名
            data: 要更新的数据字典 {列名: 值}
            where: WHERE 条件（如 "id = %s"）
            where_params: WHERE 条件的参数

        Returns:
            受影响的行数
        """
        set_clause = ', '.join([f"{k} = %s" for k in data.keys()])
        sql = f"UPDATE {table} SET {set_clause} WHERE {where}"

        params = tuple(data.values()) + (where_params or ())

        with self.get_cursor() as cursor:
            affected = cursor.execute(sql, params)
            logger.info(f"更新成功: {table}, 影响 {affected} 行")
            return affected

    def execute_delete(self, table: str, where: str, where_params: Optional[tuple] = None) -> int:
        """
        删除数据

        Args:
            table: 表名
            where: WHERE 条件（如 "id = %s"）
            where_params: WHERE 条件的参数

        Returns:
            受影响的行数
        """
        sql = f"DELETE FROM {table} WHERE {where}"

        with self.get_cursor() as cursor:
            affected = cursor.execute(sql, where_params or ())
            logger.info(f"删除成功: {table}, 影响 {affected} 行")
            return affected

    def execute_sql(self, sql: str, params: Optional[Union[tuple, dict]] = None) -> int:
        """
        执行任意 SQL 语句（DDL/DML）

        Args:
            sql: SQL 语句
            params: 参数

        Returns:
            受影响的行数
        """
        with self.get_cursor() as cursor:
            affected = cursor.execute(sql, params or ())
            logger.debug(f"SQL 执行成功，影响 {affected} 行")
            return affected

    # ==================== 事务操作 ====================

    def begin_transaction(self):
        """开始事务（需要 autocommit=False）"""
        conn = self._get_connection()
        conn.begin()
        logger.debug("事务已开始")

    def commit_transaction(self):
        """提交事务"""
        conn = self._get_connection()
        conn.commit()
        logger.debug("事务已提交")

    def rollback_transaction(self):
        """回滚事务"""
        conn = self._get_connection()
        conn.rollback()
        logger.debug("事务已回滚")

    # ==================== 辅助方法 ====================

    def table_exists(self, table_name: str) -> bool:
        """检查表是否存在"""
        sql = """
            SELECT COUNT(*) as count 
            FROM information_schema.tables 
            WHERE table_schema = %s AND table_name = %s
        """
        result = self.execute_query_one(sql, (self.config['database'], table_name))
        return result and result['count'] > 0

    def get_table_info(self, table_name: str) -> List[Dict[str, Any]]:
        """获取表结构信息"""
        sql = f"DESCRIBE {table_name}"
        return self.execute_query(sql)

    def count(self, table: str, where: Optional[str] = None, where_params: Optional[tuple] = None) -> int:
        """统计表行数"""
        sql = f"SELECT COUNT(*) as total FROM {table}"
        if where:
            sql += f" WHERE {where}"
        result = self.execute_query_one(sql, where_params or ())
        return result['total'] if result else 0


# ==================== 使用示例 ====================

if __name__ == "__main__":
    # 创建客户端实例（自动创建数据库）
    client = MySQLClientManager(
        host='127.0.0.1',
        port=3307,
        user='root',
        password='happy23',
        database='test_db',
        autocommit=False,
        auto_create_db=True  # 自动创建数据库
    )

    try:
        # 创建测试表
        create_table_sql = """
            CREATE TABLE IF NOT EXISTS users (
                id INT AUTO_INCREMENT PRIMARY KEY,
                name VARCHAR(50) NOT NULL,
                age INT,
                email VARCHAR(100),
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """
        client.execute_sql(create_table_sql)
        print("✅ 表创建成功")

        # 插入单条数据
        user_data = {'name': '张三', 'age': 25, 'email': 'zhangsan@example.com'}
        user_id = client.execute_insert('users', user_data)
        print(f"✅ 插入用户 ID: {user_id}")

        # 批量插入数据
        users = [
            {'name': '李四', 'age': 30, 'email': 'lisi@example.com'},
            {'name': '王五', 'age': 28, 'email': 'wangwu@example.com'},
            {'name': '赵六', 'age': 35, 'email': 'zhaoliu@example.com'},
        ]
        affected = client.execute_insert_batch('users', users)
        print(f"✅ 批量插入 {affected} 条记录")

        # 查询数据（年龄大于25的）
        result = client.execute_query("SELECT * FROM users WHERE age > %s", (25,))
        print(f"\n📋 年龄大于25的用户:")
        for row in result:
            print(f"  - ID: {row['id']}, 姓名: {row['name']}, 年龄: {row['age']}, 邮箱: {row['email']}")

        # 查询单条数据
        user = client.execute_query_one("SELECT * FROM users WHERE id = %s", (user_id,))
        if user:
            print(f"\n📋 ID={user_id} 的用户: {user['name']}, 年龄: {user['age']}")

        # 更新数据
        affected = client.execute_update('users', {'age': 26}, 'id = %s', (user_id,))
        print(f"✅ 更新了 {affected} 条记录")

        # 再次查询验证更新
        user = client.execute_query_one("SELECT * FROM users WHERE id = %s", (user_id,))
        if user:
            print(f"📋 更新后用户 {user['name']} 的年龄: {user['age']}")

        # 统计总记录数
        total = client.count('users')
        print(f"\n📊 用户总数: {total}")

        # 统计年龄大于30的记录数
        count_gt_30 = client.count('users', 'age > %s', (30,))
        print(f"📊 年龄大于30的用户数: {count_gt_30}")

        # 检查表是否存在
        exists = client.table_exists('users')
        print(f"\n📋 表 users 是否存在: {exists}")

        # 获取表结构
        table_info = client.get_table_info('users')
        print(f"\n📋 表结构:")
        for field in table_info:
            print(f"  - {field['Field']}: {field['Type']}")

        # 删除数据（注释掉以避免误删）
        # affected = client.execute_delete('users', 'id = %s', (user_id,))
        # print(f"✅ 删除了 {affected} 条记录")

    except Exception as e:
        print(f"❌ 操作失败: {e}")
        import traceback

        traceback.print_exc()
    finally:
        # 关闭连接
        client.close()
        print("\n✅ 数据库连接已关闭")