1、安装 Python MySQL 驱动

pip install pymysql cryptography

使用pymysql操作 MySQL 的步骤如下所示：

1. 创建连接。MySQL 服务器启动后，提供了基于 TCP （传输控制协议）的网络服务。我们可以通过pymysql模块的connect函数连接 MySQL 服务器。在调用connect函数时，需要指定主机（host）、端口（port）、用户名（user）、口令（password）、数据库（database）、字符集（charset）等参数，该函数会返回一个Connection对象。
2. 获取游标。连接 MySQL 服务器成功后，接下来要做的就是向数据库服务器发送 SQL 语句，MySQL 会执行接收到的 SQL 并将执行结果通过网络返回。要实现这项操作，需要先通过连接对象的cursor方法获取游标（Cursor）对象。
3. 发出 SQL。通过游标对象的execute方法，我们可以向数据库发出 SQL 语句。
4. 如果执行insert、delete或update操作，需要根据实际情况提交或回滚事务。因为创建连接时，默认开启了事务环境，在操作完成后，需要使用连接对象的commit或rollback方法，实现事务的提交或回滚，rollback方法通常会放在异常捕获代码块except中。如果执行select操作，需要通过游标对象抓取查询的结果，对应的方法有三个，分别是：fetchone、fetchmany和fetchall。其中fetchone方法会抓取到一条记录，并以元组或字典的方式返回；fetchmany和fetchall方法会抓取到多条记录，以嵌套元组或列表装字典的方式返回。
5. 关闭连接。在完成持久化操作后，请不要忘记关闭连接，释放外部资源。我们通常会在finally代码块中使用连接对象的close方法来关闭连接。

2、连接 MySQL 数据库

1. 创建数据库连接

import pymysql

# 建立数据库连接
connection = pymysql.connect(
    host='localhost',      # 数据库服务器地址
    user='root',      # 数据库用户名
    password='root',  # 数据库密码
    database='sql_python_demo',   # 数据库名称
    port=3306,            # 端口，MySQL默认3306
    charset='utf8mb4',     # 字符集
    cursorclass=pymysql.cursors.DictCursor  # 返回字典格式的结果
)

print("数据库连接成功!")

# 关闭连接
connection.close()

2. 使用连接池（推荐生产环境使用）

pip install dbutils

import pymysql
from dbutils.pooled_db import PooledDB

# 创建连接池
pool = PooledDB(
    creator=pymysql,
    maxconnections=10,  # 连接池最大连接数
    mincached=2,       # 初始化时创建的连接数
    host='localhost',
    user='root',
    password='root',
    database='sql_python_demo',
    port=3306,
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

# 从连接池获取连接
conn = pool.connection()
try:
    with conn.cursor() as cursor:
        cursor.execute("SELECT VERSION()")
        result = cursor.fetchone()
        print("MySQL版本:", result['VERSION()'])
finally:
    conn.close()  # 将连接返回到连接池

3、CRUD 操作示例

1. 创建表

def create_table():
    conn = pool.connection()
    try:
        with conn.cursor() as cursor:
            sql = """
            CREATE TABLE IF NOT EXISTS `users` (
                `id` INT NOT NULL AUTO_INCREMENT,
                `name` VARCHAR(50) NOT NULL,
                `email` VARCHAR(100) NOT NULL,
                `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                PRIMARY KEY (`id`),
                UNIQUE INDEX `email_UNIQUE` (`email` ASC)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
            """
            cursor.execute(sql)
        conn.commit()
        print("表创建成功")
    except Exception as e:
        conn.rollback()
        print(f"创建表失败: {e}")
    finally:
        conn.close()

create_table()

2. 插入数据

def insert_user(name, email):
    conn = pool.connection()
    try:
        with conn.cursor() as cursor:
            sql = "INSERT INTO `users` (`name`, `email`) VALUES (%s, %s)"
            cursor.execute(sql, (name, email))
        conn.commit()
        print(f"插入成功，ID: {cursor.lastrowid}")
    except pymysql.err.IntegrityError as e:
        conn.rollback()
        print(f"插入失败: 邮箱已存在")
    except Exception as e:
        conn.rollback()
        print(f"插入失败: {e}")
    finally:
        conn.close()

insert_user("张三", "zhangsan@example.com")
insert_user("李四", "lisi@example.com")

3. 查询数据

def query_users():
    conn = pool.connection()
    try:
        with conn.cursor() as cursor:
            # 查询所有用户
            cursor.execute("SELECT * FROM `users`")
            users = cursor.fetchall()
            print("所有用户:")
            for user in users:
                print(user)
            
            # 条件查询
            cursor.execute("SELECT * FROM `users` WHERE `name` LIKE %s", ('%张%',))
            zhang_users = cursor.fetchall()
            print("\n姓张的用户:")
            for user in zhang_users:
                print(user)
    except Exception as e:
        print(f"查询失败: {e}")
    finally:
        conn.close()

query_users()

4. 更新数据

def update_user(user_id, new_name):
    conn = pool.connection()
    try:
        with conn.cursor() as cursor:
            sql = "UPDATE `users` SET `name` = %s WHERE `id` = %s"
            affected_rows = cursor.execute(sql, (new_name, user_id))
            conn.commit()
            print(f"更新成功，影响行数: {affected_rows}")
    except Exception as e:
        conn.rollback()
        print(f"更新失败: {e}")
    finally:
        conn.close()

update_user(1, "张三丰")

5. 删除数据

def delete_user(user_id):
    conn = pool.connection()
    try:
        with conn.cursor() as cursor:
            sql = "DELETE FROM `users` WHERE `id` = %s"
            affected_rows = cursor.execute(sql, (user_id,))
            conn.commit()
            print(f"删除成功，影响行数: {affected_rows}")
    except Exception as e:
        conn.rollback()
        print(f"删除失败: {e}")
    finally:
        conn.close()

delete_user(2)

6、防止 SQL 注入

# 错误示范 - 容易导致SQL注入
def unsafe_query(user_input):
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            # 危险！不要这样拼接SQL
            sql = f"SELECT * FROM users WHERE name = '{user_input}'"
            cursor.execute(sql)
            return cursor.fetchall()
    finally:
        conn.close()

# 正确示范 - 使用参数化查询
def safe_query(user_input):
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            sql = "SELECT * FROM users WHERE name = %s"
            cursor.execute(sql, (user_input,))
            return cursor.fetchall()
    finally:
        conn.close()

4、完整示例：用户管理系统

import pymysql
from dbutils.pooled_db import PooledDB
from datetime import datetime
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

pool = PooledDB(
    creator=pymysql,
    maxconnections=10,  # 连接池最大连接数
    mincached=2,       # 初始化时创建的连接数
    host='localhost',
    user='root',
    password='root',
    database='sql_python_demo',
    port=3306,
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor,
    autocommit=False
)


class UserManager:
    @staticmethod
    def create_table():
        """创建用户表"""
        conn = pool.connection()
        try:
            with conn.cursor() as cursor:
                sql = """
                CREATE TABLE IF NOT EXISTS `users` (
                    `id` INT NOT NULL AUTO_INCREMENT,
                    `username` VARCHAR(50) NOT NULL,
                    `email` VARCHAR(100) NOT NULL,
                    `password_hash` VARCHAR(255) NOT NULL,
                    `is_active` BOOLEAN DEFAULT TRUE,
                    `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
                    `updated_at` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    PRIMARY KEY (`id`),
                    UNIQUE INDEX `username_UNIQUE` (`username` ASC),
                    UNIQUE INDEX `email_UNIQUE` (`email` ASC)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                """
                cursor.execute(sql)
            conn.commit()
            print("用户表创建成功")
        except Exception as e:
            conn.rollback()
            print(f"创建表失败: {e}")
        finally:
            conn.close()

    @staticmethod
    def add_user(username, email, password):
        """添加用户"""
        conn = pool.connection()
        try:
            with conn.cursor() as cursor:
                # 在实际应用中应该使用安全的密码哈希
                password_hash = f"hashed_{password}"  # 这里应该使用bcrypt等库

                sql = """
                INSERT INTO `users` (`username`, `email`, `password_hash`)
                VALUES (%s, %s, %s)
                """
                cursor.execute(sql, (username, email, password_hash))
            conn.commit()
            print(f"用户添加成功，ID: {cursor.lastrowid}")
            return cursor.lastrowid
        except pymysql.err.IntegrityError as e:
            conn.rollback()
            if "username" in str(e):
                raise ValueError("用户名已存在")
            elif "email" in str(e):
                raise ValueError("邮箱已存在")
            else:
                raise ValueError("添加用户失败")
        except Exception as e:
            conn.rollback()
            raise ValueError(f"添加用户失败: {e}")
        finally:
            conn.close()

    @staticmethod
    def get_user(user_id=None, username=None, email=None):
        """获取用户信息"""
        conn = pool.connection()
        try:
            with conn.cursor() as cursor:
                if user_id:
                    sql = "SELECT * FROM `users` WHERE `id` = %s"
                    cursor.execute(sql, (user_id,))
                elif username:
                    sql = "SELECT * FROM `users` WHERE `username` = %s"
                    cursor.execute(sql, (username,))
                elif email:
                    sql = "SELECT * FROM `users` WHERE `email` = %s"
                    cursor.execute(sql, (email,))
                else:
                    raise ValueError("必须提供user_id、username或email之一")

                user = cursor.fetchone()
                if not user:
                    raise ValueError("用户不存在")
                return user
        except Exception as e:
            raise ValueError(f"获取用户失败: {e}")
        finally:
            conn.close()

    @staticmethod
    def update_user(user_id, **kwargs):
        """更新用户信息"""
        if not kwargs:
            return

        conn = pool.connection()
        try:
            with conn.cursor() as cursor:
                set_clause = ", ".join([f"`{k}` = %s" for k in kwargs])
                sql = f"UPDATE `users` SET {set_clause} WHERE `id` = %s"
                values = list(kwargs.values()) + [user_id]
                affected_rows = cursor.execute(sql, values)

                if affected_rows == 0:
                    raise ValueError("用户不存在或数据未更改")
            conn.commit()
            print(f"用户信息更新成功，影响行数: {affected_rows}")
        except pymysql.err.IntegrityError as e:
            conn.rollback()
            if "username" in str(e):
                raise ValueError("用户名已存在")
            elif "email" in str(e):
                raise ValueError("邮箱已存在")
            else:
                raise ValueError("更新用户失败")
        except Exception as e:
            conn.rollback()
            raise ValueError(f"更新用户失败: {e}")
        finally:
            conn.close()

    @staticmethod
    def delete_user(user_id):
        """删除用户（标记为不活跃）"""
        conn = pool.connection()
        try:
            with conn.cursor() as cursor:
                sql = "UPDATE `users` SET `is_active` = FALSE WHERE `id` = %s"
                affected_rows = cursor.execute(sql, (user_id,))

                if affected_rows == 0:
                    raise ValueError("用户不存在")
            conn.commit()
            print(f"用户删除成功，影响行数: {affected_rows}")
        except Exception as e:
            conn.rollback()
            raise ValueError(f"删除用户失败: {e}")
        finally:
            conn.close()

    @staticmethod
    def list_users(active_only=True, page=1, per_page=10):
        """分页列出用户"""
        conn = pool.connection()
        try:
            with conn.cursor() as cursor:
                where = "WHERE `is_active` = TRUE" if active_only else ""
                sql = f"""
                SELECT * FROM `users` {where}
                ORDER BY `id` DESC
                LIMIT %s OFFSET %s
                """
                offset = (page - 1) * per_page
                cursor.execute(sql, (per_page, offset))
                users = cursor.fetchall()

                # 获取总数
                count_sql = f"SELECT COUNT(*) as total FROM `users` {where}"
                cursor.execute(count_sql)
                total = cursor.fetchone()['total']

                return {
                    'users': users,
                    'total': total,
                    'page': page,
                    'per_page': per_page,
                    'total_pages': (total + per_page - 1) // per_page
                }
        except Exception as e:
            raise ValueError(f"获取用户列表失败: {e}")
        finally:
            conn.close()


# 使用示例
if __name__ == "__main__":
    try:
        UserManager.create_table()

        # 添加用户
        user_id = UserManager.add_user("testuser", "test@example.com", "password123")
        print(f"添加的用户ID: {user_id}")

        # 获取用户
        user = UserManager.get_user(user_id=user_id)
        print("用户信息:", user)

        # 更新用户
        UserManager.update_user(user_id, email="newemail@example.com")
        updated_user = UserManager.get_user(user_id=user_id)
        print("更新后的用户:", updated_user)

        # 分页列出用户
        result = UserManager.list_users()
        print("用户列表:", result)

        # 删除用户
        UserManager.delete_user(user_id)
        deleted_user = UserManager.get_user(user_id=user_id)
        print("删除后的用户状态:", deleted_user['is_active'])

    except ValueError as e:
        print(f"操作失败: {e}")