Python使用PyMySQL连接MySQL数据库及基础增删改查示例

在 Python 应用程序中与 MySQL 数据库进行通信，最基础也最常用的方案是使用纯 Python 实现的驱动客户端 PyMySQL。它的核心工作逻辑可以概括为四个环节：通过 TCP 套接字与 MySQL 服务端建立连接（Connection）、申请用于执行 SQL 指令的游标对象（Cursor）、利用参数化查询执行增删改查并维持事务一致性、在业务处理结束后完整释放连接资源。本文将以可直接运行的代码示例为核心，完整讲解连接建立、表结构准备、增删改查（CRUD）操作、防 SQL 注入机制以及健壮的异常闭环处理方法。

一、环境准备与建立数据库连接

PyMySQL 是一个符合 Python DB API v2.0 规范的纯 Python 客户端库，无需在操作系统中编译或安装 MySQL 原生 C 客户端依赖（如 mysqlclient 所需的本地头文件），因此在各种操作系统和容器环境中都能直接安装使用。

1.1 安装 PyMySQL

在当前 Python 虚拟环境中使用 pip 命令进行安装：

pip install pymysql

如果业务中需要支持 MySQL 8.0 默认的 caching_sha2_password 身份验证插件，建议同时安装密码学支持模块 cryptography：

pip install cryptography

1.2 建立第一个连接：pymysql.connect 参数说明

连接 MySQL 数据库通过调用 pymysql.connect() 函数实现。该函数会向目标主机发起 TCP 握手并完成认证，返回一个 Connection 连接对象。以下是建立连接的标准示例代码：

import pymysql

# 建立 MySQL 数据库连接
connection = pymysql.connect(
    host='127.0.0.1',        # 数据库主机地址（本地使用 127.0.0.1 或 localhost）
    port=3306,               # MySQL 服务端口（默认 3306）
    user='app_user',         # 数据库用户名
    password='SecurePassword123!', # 数据库密码
    database='shop_db',       # 目标数据库名
    charset='utf8mb4',       # 字符集编码，务必使用 utf8mb4
    cursorclass=pymysql.cursors.DictCursor,  # 游标返回字典格式
    connect_timeout=10       # 连接超时时间（秒）
)

print("数据库连接建立成功，当前连接状态:", connection.open)
connection.close()

二、理解连接（Connection）与游标（Cursor）的分工

很多初学者容易将连接与游标混为一谈，甚至直接尝试调用 conn.execute()，这会直接导致属性错误。在 PyMySQL 的设计中，连接（Connection）与游标（Cursor）有着严格的职责划分：

• Connection（连接对象）：代表应用程序与 MySQL 服务端之间的一条物理 TCP 网络通道。它负责握手鉴权、会话参数协商，以及最重要的事务边界控制（调用 conn.commit() 提交事务或 conn.rollback() 回滚事务）。
• Cursor（游标对象）：依托于具体的连接对象创建（cursor = conn.cursor()）。它负责真正发送 SQL 指令给数据库服务器解析执行，维护操作结果的状态信息（例如受影响行数 rowcount、自增主键值 lastrowid），并在客户端内部维持一个用于逐行读取结果集的指针。

2.1 游标模式辨析：默认 Cursor 与 DictCursor

PyMySQL 提供了多种游标类型，最核心的两种是默认游标与字典游标：

1. 默认元组游标（pymysql.cursors.Cursor）：如果不做指定，执行查询返回的每一行记录都是一个 Python 元组（tuple），数据按 SQL 查询列的顺序排列。访问字段必须使用数字下标，例如 row[0]、row[1]。这种模式内存开销极小，但在业务开发中可读性较差，一旦数据表新增字段或修改了 SELECT 列顺序，数字下标极易产生错位 bug。
2. 字典游标（pymysql.cursors.DictCursor）：在建立连接或创建游标时指定 cursorclass=pymysql.cursors.DictCursor，每一行查询结果都会被转换为一个以字段名为键、字段值为对应值的 Python 字典（dict），例如 row['username']。这种方式不仅代码意图清晰，还能直接配合 json.dumps() 序列化输出给前端或 API 接口，是绝大多数业务场景下的推荐选择。

三、基础增删改查（CRUD）操作完整实现

为了保证示例代码的前后一致与可重现性，我们先准备一张用户示例表 users：

-- 准备演示数据表
CREATE DATABASE IF NOT EXISTS shop_db DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE shop_db;

CREATE TABLE IF NOT EXISTS users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    username VARCHAR(50) NOT NULL UNIQUE,
    email VARCHAR(100) NOT NULL,
    age INT DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

3.1 新增数据（Create）：单条插入与批量插入

执行写操作时，有两个关键规则必须遵守：一是使用 %s 作为参数化占位符，二是写操作完成后必须显式调用 connection.commit()。

import pymysql
from pymysql.cursors import DictCursor

conn = pymysql.connect(
    host='127.0.0.1', port=3306, user='app_user',
    password='SecurePassword123!', database='shop_db',
    charset='utf8mb4', cursorclass=DictCursor
)

try:
    with conn.cursor() as cursor:
        # 单条插入：使用 %s 参数化占位符
        insert_sql = """
            INSERT INTO users (username, email, age)
            VALUES (%s, %s, %s);
        """
        # 参数使用元组传递
        cursor.execute(insert_sql, ('zhangsan', 'zhangsan@example.com', 25))
        
        # 获取自增 ID
        new_id = cursor.lastrowid
        print(f"单条数据插入成功，生成的主键 ID: {new_id}，受影响行数: {cursor.rowcount}")

        # 批量插入：使用 executemany
        batch_data = [
            ('lisi', 'lisi@example.com', 28),
            ('wangwu', 'wangwu@example.com', 32),
            ('zhaoliu', 'zhaoliu@example.com', 22)
        ]
        cursor.executemany(insert_sql, batch_data)
        print(f"批量插入完成，受影响行数: {cursor.rowcount}")

    # 提交事务：将修改永久保存到磁盘
    conn.commit()
    print("事务提交成功，数据已成功持久化。")

except Exception as e:
    conn.rollback()
    print("写入过程发生异常，已回滚事务:", e)
finally:
    conn.close()

3.2 查询数据（Read）：获取单条、全部及分页数据

查询操作不会改变表内数据，因此不需要执行 commit()。执行查询主要通过游标的提取方法读取数据：

• cursor.fetchone()：从结果集中读取下一行数据。如果没有更多记录，返回 None。
• cursor.fetchall()：将匹配到的所有剩余行一次性读取为列表。若数据量过大（如数十万行），会占用大量客户端内存，应配合 SQL LIMIT 分页使用。
• cursor.fetchmany(size=N)：每次读取指定数量 N 的记录，适合做流式批处理。

import pymysql
from pymysql.cursors import DictCursor

conn = pymysql.connect(
    host='127.0.0.1', port=3306, user='app_user',
    password='SecurePassword123!', database='shop_db',
    charset='utf8mb4', cursorclass=DictCursor
)

try:
    with conn.cursor() as cursor:
        # 查询单条记录
        query_one_sql = "SELECT id, username, email, age FROM users WHERE username = %s;"
        cursor.execute(query_one_sql, ('zhangsan',))
        user = cursor.fetchone()
        
        if user:
            print(f"找到用户: 用户名={user['username']}, 邮箱={user['email']}, 年龄={user['age']}")
        else:
            print("未找到指定用户")

        # 查询多条记录（附带范围过滤与排序）
        query_all_sql = """
            SELECT id, username, email, age 
            FROM users 
            WHERE age >= %s 
            ORDER BY age DESC 
            LIMIT 10;
        """
        cursor.execute(query_all_sql, (24,))
        user_list = cursor.fetchall()
        
        print(f"符合条件的用户总数: {len(user_list)}")
        for idx, item in enumerate(user_list, start=1):
            print(f"  [{idx}] ID: {item['id']} | 姓名: {item['username']} | 年龄: {item['age']}")

finally:
    conn.close()

3.3 更新数据（Update）：基于受影响行数判断结果

执行数据修改时，SQL 语句使用参数化传入更新的值与检索条件，并通过 cursor.rowcount 获取实际被修改的行数：

import pymysql

conn = pymysql.connect(
    host='127.0.0.1', port=3306, user='app_user',
    password='SecurePassword123!', database='shop_db',
    charset='utf8mb4'
)

try:
    with conn.cursor() as cursor:
        update_sql = """
            UPDATE users 
            SET age = %s, email = %s 
            WHERE username = %s;
        """
        # 注意顺序与占位符严格匹配
        params = (26, 'zhangsan_new@example.com', 'zhangsan')
        affected_rows = cursor.execute(update_sql, params)
        print(f"执行更新完毕，受影响行数: {affected_rows}")

    conn.commit()
    print("更新已成功提交并落盘。")
except Exception as e:
    conn.rollback()
    print("更新失败，事务已回滚:", e)
finally:
    conn.close()

3.4 删除数据（Delete）：条件限定与受影响判定

删除操作在业务中通常具有高风险性，必须严格指定 WHERE 条件，防止全表误删：

import pymysql

conn = pymysql.connect(
    host='127.0.0.1', port=3306, user='app_user',
    password='SecurePassword123!', database='shop_db',
    charset='utf8mb4'
)

try:
    with conn.cursor() as cursor:
        delete_sql = "DELETE FROM users WHERE username = %s;"
        cursor.execute(delete_sql, ('zhaoliu',))
        
        if cursor.rowcount > 0:
            print(f"成功删除 {cursor.rowcount} 条数据")
        else:
            print("未找到匹配条件的数据，未执行任何删除")

    conn.commit()
except Exception as e:
    conn.rollback()
    print("删除异常，已执行回滚:", e)
finally:
    conn.close()

四、参数化查询与 SQL 注入防范机制

在编写数据库交互代码时，最严重的开发事故往往来自于直接拼装 SQL 语句。

4.1 危险做法与常见误解

很多开发者在初学时习惯使用 Python 的字符串拼接或 f-string：

# 严重高危代码！绝对不要在生产环境使用！
user_input = "' OR '1'='1"
unsafe_sql = f"SELECT * FROM users WHERE username = '{user_input}';"
cursor.execute(unsafe_sql)
# 拼接出的完整 SQL 变为: SELECT * FROM users WHERE username = '' OR '1'='1';
# 这将绕过认证并导致整张表的数据全部被非法提取！

4.2 PyMySQL 参数化查询的内部工作原理

PyMySQL 采用 %s 作为参数占位符（无论字段是字符串、整数、浮点数还是日期类型，占位符统一使用 %s，且占位符外绝对不能加单引号或双引号）：

# 正确规范：占位符不加引号，参数打包为元组传入
safe_sql = "SELECT * FROM users WHERE username = %s;"
cursor.execute(safe_sql, (user_input,))

当执行上述语句时，PyMySQL 内部会调用其自带的转义函数（escape_args），自动识别参数的数据类型。如果传入字符串，驱动会自动为该字符串包裹闭合引号，并对内部包含的特殊字符（如单引号、反斜杠、空字节等）做严格的转义过滤。最终送到 MySQL 服务器引擎时，传入的输入只能作为纯文本字面量（Literal Value）参与匹配，绝对无法改变原 SQL 语句的抽象语法树（AST），从而在底层彻底根除了 SQL 注入的可能。

五、生产级异常处理与资源释放最佳实践

数据库连接是非常宝贵的网络与系统资源，MySQL 实例对最大并发连接数（max_connections）有严格上限。如果代码抛出异常时未能妥善关闭连接，未释放的连接会持续占用端口和内存句柄，最终导致数据库报出 Too many connections 错误，使整个服务陷入瘫痪。

5.1 健壮的工程模板代码

在生产业务模块中，推荐将数据库交互逻辑封装为带有严格异常捕获与资源清理的闭环结构：

import pymysql
from pymysql.cursors import DictCursor

def update_user_email(user_id: int, new_email: str) -> bool:
    """
    修改指定用户的邮箱地址，返回修改是否成功的布尔值
    """
    conn = None
    try:
        # 1. 建立连接
        conn = pymysql.connect(
            host='127.0.0.1',
            port=3306,
            user='app_user',
            password='SecurePassword123!',
            database='shop_db',
            charset='utf8mb4',
            cursorclass=DictCursor,
            connect_timeout=5
        )

        # 2. 使用上下文管理器自动释放 cursor
        with conn.cursor() as cursor:
            sql = "UPDATE users SET email = %s WHERE id = %s;"
            cursor.execute(sql, (new_email, user_id))
            
            # 3. 提交写操作事务
            conn.commit()
            print(f"用户 ID {user_id} 的邮箱已更新为 {new_email}")
            return cursor.rowcount > 0

    except pymysql.MySQLError as db_err:
        # 4. 捕获数据库相关异常并执行回滚
        if conn:
            conn.rollback()
        # 提取错误码与错误信息记录日志
        error_code, error_msg = db_err.args[0], db_err.args[1] if len(db_err.args) > 1 else ""
        print(f"数据库操作失败! 错误码: {error_code}, 详情: {error_msg}")
        return False

    except Exception as ex:
        if conn:
            conn.rollback()
        print(f"未知业务异常: {ex}")
        return False

    finally:
        # 5. 确保底层连接在任何情况下都能被关闭
        if conn and conn.open:
            conn.close()
            print("数据库连接已安全关闭。")

# 测试调用
if __name__ == '__main__':
    result = update_user_email(1, "zhangsan_verified@example.com")
    print(f"操作最终状态: {result}")

六、PyMySQL 常见错误及排查清单

七、总结与核心注意事项

通过 PyMySQL 操作 MySQL 数据库是一套标准且严谨的工程过程。在日常开发与项目上线前，建议对照以下四条核心准则进行代码自检：

• 字符集统一规范：始终在连接配置与建表语句中指定 utf8mb4，防止特殊符号与 Emoji 导致存取失败。
• 写操作显式提交：牢记单条或批量执行 INSERT、UPDATE、DELETE 之后，必须调用 conn.commit()，否则修改仅保留于会话缓存并在断开连接时隐式回滚。
• 绝对防范 SQL 注入：任何来自外部用户输入的动态数据，均必须使用 %s 占位符并通过元组传入 execute，严禁使用 f-string、+ 或 % 进行字符串拼接。
• 保证资源闭环清理：充分利用 with conn.cursor() 上下文管理器及 try...finally 结构，确保游标与连接在所有分支下都能被及时释放。若业务为高并发 API 服务，应进一步引入连接池（如 DBUtils.PooledDB）以降低高频建立 TCP 连接的性能开销。