Python使用PyMySQL连接MySQL数据库的完整步骤

在 Python 后端开发与数据处理场景中，PyMySQL 是最为直接、轻量且被官方生态广泛推荐的 MySQL 驱动库。作为使用纯 Python 实现的数据库驱动，PyMySQL 无需在操作系统中编译 C 语言底层扩展或依赖特定 MySQL 客户端库，具备出色的跨平台兼容性与安装便捷性。要通过 PyMySQL 完成对 MySQL 数据库的稳健访问，核心流程由六个关键环节组成：准备开发环境并安装依赖、配置网络参数建立 Connection 连接对象、创建数据交互的 Cursor 游标、执行参数化 SQL 语句处理数据、严格管理写操作的事务提交与异常回滚，以及在操作完成后安全释放资源。本文按照实际工程操作步骤展开，逐一讲解具体实现细节与关键注意事项。

一、准备环境与安装 PyMySQL 依赖

在编写连接代码前，需要确保本地或服务器的 Python 环境中已就绪 PyMySQL 模块。PyMySQL 支持 Python 3.7 及以上版本，可以通过标准 pip 工具直接安装。

打开终端并执行以下安装命令：

pip install pymysql

安装完成后，可以在 Python 交互式环境中通过简单的导入验证是否安装成功：

# 验证安装的测试代码
import pymysql

print("PyMySQL 版本号：", pymysql.__version__)

若终端正常打印出版本号且未提示 ModuleNotFoundError，则表示驱动环境已准备就绪。

二、配置连接参数并建立数据库连接

连接 MySQL 数据库的核心是调用 pymysql.connect() 方法。该方法会在客户端与远程或本地的 MySQL 实例之间建立 TCP 套接字通道，并完成身份认证与字符集协商，成功后返回一个 Connection 实例。

连接方法中常用的关键参数及其作用如下：

以下是建立连接的完整基础代码示例：

import pymysql
import pymysql.cursors

# 建立数据库连接对象
connection = pymysql.connect(
    host='127.0.0.1',
    user='root',
    password='your_secure_password',
    database='shop_db',
    port=3306,
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor,
    connect_timeout=10
)

print("数据库连接成功，连接状态：", connection.open)

三、创建游标并执行数据查询（SELECT）

建立了 Connection 对象后，并不能直接发送 SQL 语句，必须通过连接对象创建 Cursor（游标）。游标承担着两个关键职责：将 SQL 指令发送给数据库服务器执行，以及在客户端接收并遍历结果集缓存。

1. 获取游标与参数化查询

在实际业务中执行查询时，强烈要求使用参数化传递方式传递动态变量，严禁使用 Python 的字符串拼接或 f-string。PyMySQL 会在底层自动对传入的参数进行安全转义，彻底杜绝 SQL 注入漏洞。

# 创建游标
cursor = connection.cursor()

# 编写参数化 SQL 模板，统一使用 %s 占位符（无论字段类型是整数、字符串还是日期）
sql = "SELECT id, username, email, balance FROM users WHERE status = %s AND balance >= %s"
params = ('active', 100.0)

# 执行 SQL 语句
cursor.execute(sql, params)

2. 获取查询结果集

cursor.execute() 执行后返回的是匹配的行数整数，真正的数据行需要通过以下三种方法之一从游标缓冲区中提取：

• cursor.fetchone()：提取结果集中的下一行。若已到末尾则返回 None。适用于已知只返回单条记录的查询或内存敏感的大数据流式处理。
• cursor.fetchmany(size)：按指定的批量大小提取多行记录，返回列表。常用于分批分页拉取数据。
• cursor.fetchall()：一次性提取查询匹配的全部记录。返回包含字典（或元组）的列表。若数据量过大（如数十万行），会显著占用 Python 进程内存。

# 示例：遍历并打印查询到的字典格式数据
rows = cursor.fetchall()
for row in rows:
    print(f"用户ID: {row['id']}, 用户名: {row['username']}, 余额: {row['balance']}")

四、执行数据变更与事务管理（INSERT / UPDATE / DELETE）

对于数据的增、删、改等写操作，MySQL 的常用存储引擎（如 InnoDB）采用事务机制。在默认配置下，PyMySQL 建立的连接处于 autocommit=False（非自动提交）状态。这意味着执行完 cursor.execute() 后，更改仅处于当前会话的事务暂存区中，如果进程退出或发生异常，更改将被自动丢弃，必须显式调用 connection.commit() 才能将更改持久化到磁盘。

1. 单条记录插入与获取自增主键

insert_sql = "INSERT INTO users (username, email, balance) VALUES (%s, %s, %s)"
data = ("zhangsan", "zhangsan@example.com", 250.0)

try:
    cursor.execute(insert_sql, data)
    # 获取刚插入记录的自增 ID
    new_user_id = cursor.lastrowid
    print(f"新用户添加成功，自增主键ID: {new_user_id}")
    
    # 显式提交事务
    connection.commit()
except Exception as e:
    # 发生异常立即回滚，保证数据库状态一致
    connection.rollback()
    print(f"数据插入失败，已触发回滚：{e}")

2. 批量插入高效优化：executemany

若需要同时插入多条数据，连续调用 cursor.execute() 会产生频繁的网络往返开销。此时应使用 cursor.executemany() 方法，该方法会在底层将多条数据合并为批量 SQL 传输，大幅提升入库吞吐率：

batch_users = [
    ("lisi", "lisi@example.com", 180.0),
    ("wangwu", "wangwu@example.com", 320.0),
    ("zhaoliu", "zhaoliu@example.com", 95.0)
]

try:
    affected_rows = cursor.executemany(insert_sql, batch_users)
    connection.commit()
    print(f"批量插入完成，累计影响行数: {affected_rows}")
except Exception as e:
    connection.rollback()
    print(f"批量插入失败，事务已回滚：{e}")

五、安全释放资源与上下文管理器模式

数据库连接和游标属于宝贵的系统网络与内存资源。如果代码发生异常或连接长期闲置未关闭，会导致 MySQL 服务器的并发连接池被逐渐耗尽（报 Too many connections 错误）。因此，每一个打开的游标与连接都必须在操作完成后执行关闭操作。

1. 传统标准的 try ... finally 资源回收

import pymysql

connection = None
cursor = None
try:
    connection = pymysql.connect(
        host='127.0.0.1',
        user='root',
        password='your_secure_password',
        database='shop_db',
        charset='utf8mb4'
    )
    cursor = connection.cursor()
    cursor.execute("SELECT 1")
    print(cursor.fetchone())
except pymysql.MySQLError as err:
    print(f"数据库操作异常: {err}")
finally:
    # 按照先创建后关闭原则，先关游标再关连接
    if cursor:
        cursor.close()
    if connection and connection.open:
        connection.close()
        print("数据库连接已安全关闭")

2. 现代优雅写法：使用 with 上下文管理器

PyMySQL 原生支持 Python 的 with 上下文协议。将游标纳入 with connection.cursor() as cursor: 块中管理，当代码块执行完毕或抛出未捕获异常时，Python 会自动调用 cursor.close()，避免因遗忘清理代码而引发资源泄漏。

import pymysql
import pymysql.cursors

config = {
    'host': '127.0.0.1',
    'user': 'root',
    'password': 'your_secure_password',
    'database': 'shop_db',
    'charset': 'utf8mb4',
    'cursorclass': pymysql.cursors.DictCursor
}

# 连接对象管理
conn = pymysql.connect(**config)
try:
    # 游标对象通过 with 自动清理
    with conn.cursor() as cursor:
        cursor.execute("SELECT COUNT(*) AS total_count FROM users")
        result = cursor.fetchone()
        print("用户总数：", result['total_count'])
finally:
    conn.close()

六、常见连接报错与快速排查

初次连接 MySQL 时，如果遇到错误，可以对照以下典型错误码进行针对性排查：

• pymysql.err.OperationalError: (2003, "Can't connect to MySQL server")：
  原因多为 MySQL 服务未启动、主机地址或端口填写错误、或者服务器防火墙阻止了 3306 端口的网络入站请求。请检查 systemctl status mysqld 及安全组规则。
• pymysql.err.OperationalError: (1045, "Access denied for user...")：
  用户名或密码错误，或者是该账号未开启当前客户端 IP 的访问权限（MySQL 默认用户通常绑定 'root'@'localhost'，远程连接需显式授权 'root'@'%'）。
• pymysql.err.InternalError: (1366, "Incorrect string value...")：
  字符集冲突引起。通常是 Python 连接未指定 charset='utf8mb4'，或者数据表字段的字符集仍为不支持 4 字节的旧 utf8（utf8mb3）。
• pymysql.err.OperationalError: (2006, "MySQL server has gone away")：
  连接在空闲状态下超过了 MySQL 服务端的 wait_timeout 被单方面关闭。在常驻后台任务中，执行操作前应通过 connection.ping(reconnect=True) 检查或使用连接池重连。

七、完整可执行代码范例

综合上述各个步骤，以下是一份组织严谨、具备完整异常捕获与事务回滚的端到端可执行脚本：

import pymysql
import pymysql.cursors

def run_database_operations():
    # 1. 配置参数
    db_config = {
        'host': '127.0.0.1',
        'port': 3306,
        'user': 'root',
        'password': 'your_password',
        'database': 'test_db',
        'charset': 'utf8mb4',
        'cursorclass': pymysql.cursors.DictCursor
    }

    connection = None
    try:
        # 2. 建立连接
        print("正在连接数据库...")
        connection = pymysql.connect(**db_config)
        print("连接成功！")

        # 3. 使用游标执行 DDL 和 DML
        with connection.cursor() as cursor:
            # 创建临时演示表
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS demo_items (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    item_name VARCHAR(100) NOT NULL,
                    price DECIMAL(10,2) NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
            """)

            # 4. 执行写入（参数化防注入）
            insert_query = "INSERT INTO demo_items (item_name, price) VALUES (%s, %s)"
            cursor.execute(insert_query, ("机械键盘", 499.00))
            new_id = cursor.lastrowid
            print(f"数据写入完成，生成的记录ID为: {new_id}")

            # 5. 显式提交事务
            connection.commit()
            print("事务提交成功！")

            # 6. 查询刚写入的数据
            select_query = "SELECT * FROM demo_items WHERE id = %s"
            cursor.execute(select_query, (new_id,))
            record = cursor.fetchone()
            print("查询结果详情：", record)

    except pymysql.MySQLError as error:
        # 捕获数据库相关异常并回滚
        if connection:
            connection.rollback()
        print(f"数据库操作发生错误，已触发回滚：{error}")
    finally:
        # 7. 安全释放连接
        if connection and connection.open:
            connection.close()
            print("数据库连接已关闭。")

if __name__ == '__main__':
    run_database_operations()