Python连接MySQL实战：PyMySQL基础与CRUD操作

1. PyMySQL基础与环境准备

PyMySQL是Python中连接MySQL数据库最常用的纯Python驱动之一，它完全遵循Python DB-API 2.0规范（PEP 249）。与MySQL官方的Connector/Python相比，PyMySQL的优势在于它不需要任何外部依赖，完全用Python实现，这使得它在各种平台上都能轻松安装和使用。

1.1 安装PyMySQL

安装PyMySQL非常简单，使用pip命令即可完成：

pip install PyMySQL

如果你需要使用更安全的认证方式，比如MySQL 8.0默认的caching_sha2_password插件，或者MariaDB的ed25519认证方法，可以安装额外的依赖：

pip install PyMySQL[rsa] # 支持sha256_password和caching_sha2_password pip install PyMySQL[ed25519] # 支持MariaDB的ed25519认证

注意：PyMySQL 1.0+版本要求Python 3.9或更高版本。如果你使用的是较旧的Python版本，需要安装PyMySQL 0.10.x系列。

1.2 基本连接配置

建立数据库连接是使用PyMySQL的第一步，下面是一个最基本的连接示例：

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

连接参数说明：

• host: MySQL服务器地址，可以是IP或域名
• user: 数据库用户名
• password: 用户密码
• database: 要连接的数据库名
• port: MySQL服务端口，默认3306
• charset: 字符集，推荐使用utf8mb4以支持完整的Unicode字符
• cursorclass: 游标类型，DictCursor会返回字典形式的结果

1.3 连接池管理

在高并发应用中，频繁创建和关闭连接会影响性能。PyMySQL本身不提供连接池功能，但可以通过第三方库如DBUtils来实现：

from dbutils.pooled_db import PooledDB import pymysql # 创建连接池 pool = PooledDB( creator=pymysql, maxconnections=20, # 连接池最大连接数 mincached=5, # 初始化时创建的连接数 host='localhost', user='user', password='pass', database='test', charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) # 从连接池获取连接 connection = pool.connection()

使用连接池时，获取的连接在使用完毕后需要显式关闭，否则会导致连接泄漏：

try: with connection.cursor() as cursor: cursor.execute("SELECT * FROM users") result = cursor.fetchall() finally: connection.close() # 将连接返回到连接池

2. 基本CRUD操作

2.1 创建表

在操作数据之前，通常需要先创建表结构。下面是一个创建用户表的示例：

def create_table(): connection = pymysql.connect(host='localhost', user='user', password='pass', database='test') try: with connection.cursor() as cursor: sql = """ CREATE TABLE IF NOT EXISTS `users` ( `id` INT(11) NOT NULL AUTO_INCREMENT, `name` VARCHAR(255) NOT NULL, `email` VARCHAR(255) NOT NULL, `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (`id`), UNIQUE KEY `email` (`email`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ cursor.execute(sql) connection.commit() finally: connection.close()

表设计注意事项：

1. 使用InnoDB引擎，它支持事务和外键
2. 字符集使用utf8mb4以支持完整的Unicode字符（包括emoji）
3. 为常用查询字段添加索引
4. 为必填字段设置NOT NULL约束
5. 为唯一性字段添加UNIQUE约束

2.2 插入数据

插入数据是最基本的操作之一，PyMySQL支持单条插入和批量插入：

def insert_user(name, email): connection = pymysql.connect(host='localhost', user='user', password='pass', database='test') try: with connection.cursor() as cursor: # 单条插入 sql = "INSERT INTO `users` (`name`, `email`) VALUES (%s, %s)" cursor.execute(sql, (name, email)) connection.commit() finally: connection.close() def batch_insert_users(user_list): connection = pymysql.connect(host='localhost', user='user', password='pass', database='test') try: with connection.cursor() as cursor: # 批量插入 sql = "INSERT INTO `users` (`name`, `email`) VALUES (%s, %s)" cursor.executemany(sql, user_list) connection.commit() finally: connection.close()

重要提示：永远使用参数化查询（%s占位符），而不是字符串拼接，以防止SQL注入攻击。

2.3 查询数据

PyMySQL提供了多种查询数据的方法：

def get_users(): connection = pymysql.connect(host='localhost', user='user', password='pass', database='test') try: with connection.cursor() as cursor: # 查询所有用户 cursor.execute("SELECT * FROM `users`") result = cursor.fetchall() # 获取所有记录 for row in result: print(row) # 查询单个用户 cursor.execute("SELECT * FROM `users` WHERE `id` = %s", (1,)) result = cursor.fetchone() # 获取单条记录 print(result) # 分页查询 cursor.execute("SELECT * FROM `users` LIMIT %s OFFSET %s", (10, 0)) result = cursor.fetchmany(10) # 获取指定数量的记录 print(result) finally: connection.close()

2.4 更新和删除数据

更新和删除操作与插入类似，但需要特别注意WHERE条件，避免误操作：

def update_user(user_id, new_name): connection = pymysql.connect(host='localhost', user='user', password='pass', database='test') try: with connection.cursor() as cursor: sql = "UPDATE `users` SET `name` = %s WHERE `id` = %s" affected_rows = cursor.execute(sql, (new_name, user_id)) print(f"更新了{affected_rows}条记录") connection.commit() finally: connection.close() def delete_user(user_id): connection = pymysql.connect(host='localhost', user='user', password='pass', database='test') try: with connection.cursor() as cursor: sql = "DELETE FROM `users` WHERE `id` = %s" affected_rows = cursor.execute(sql, (user_id,)) print(f"删除了{affected_rows}条记录") connection.commit() finally: connection.close()

3. 高级特性与性能优化

3.1 事务管理

MySQL的InnoDB引擎支持事务，PyMySQL也提供了完整的事务支持：

def transfer_money(from_id, to_id, amount): connection = pymysql.connect(host='localhost', user='user', password='pass', database='test') try: with connection.cursor() as cursor: # 开始事务 connection.begin() try: # 扣款 cursor.execute("UPDATE `accounts` SET `balance` = `balance` - %s WHERE `id` = %s AND `balance` >= %s", (amount, from_id, amount)) if cursor.rowcount == 0: raise Exception("扣款失败，余额不足或账户不存在") # 收款 cursor.execute("UPDATE `accounts` SET `balance` = `balance` + %s WHERE `id` = %s", (amount, to_id)) if cursor.rowcount == 0: raise Exception("收款账户不存在") # 提交事务 connection.commit() print("转账成功") except Exception as e: # 回滚事务 connection.rollback() print(f"转账失败: {str(e)}") finally: connection.close()

事务使用要点：

1. 明确调用begin()开始事务
2. 在try块中执行所有操作
3. 成功时调用commit()
4. 失败时调用rollback()
5. 确保在finally块中关闭连接

3.2 批量操作优化

对于大量数据的插入或更新，批量操作可以显著提高性能：

def bulk_insert_users(user_data): connection = pymysql.connect(host='localhost', user='user', password='pass', database='test') try: with connection.cursor() as cursor: # 开启事务 connection.begin() # 批量插入 sql = """INSERT INTO `users` (`name`, `email`, `created_at`) VALUES (%s, %s, %s)""" cursor.executemany(sql, user_data) # 提交事务 connection.commit() print(f"成功插入{cursor.rowcount}条记录") except Exception as e: connection.rollback() print(f"批量插入失败: {str(e)}") finally: connection.close()

批量操作优化技巧：

1. 使用executemany()代替循环执行execute()
2. 合理设置批量操作的大小（通常1000-5000条记录一批）
3. 在批量操作中使用事务
4. 考虑使用LOAD DATA INFILE对于超大数据量

3.3 预处理语句

预处理语句可以提高性能并防止SQL注入：

def get_user_by_id(user_id): connection = pymysql.connect(host='localhost', user='user', password='pass', database='test') try: with connection.cursor() as cursor: # 创建预处理语句 stmt = "SELECT * FROM `users` WHERE `id` = %s" # 执行查询 cursor.execute(stmt, (user_id,)) result = cursor.fetchone() print(result) finally: connection.close()

预处理语句的优势：

1. 数据库只需解析SQL一次
2. 自动处理参数转义，防止SQL注入
3. 对于重复执行的查询性能更好

4. 常见问题与解决方案

4.1 连接超时问题

MySQL服务器默认会在8小时不活动后关闭连接，这会导致PyMySQL抛出"MySQL server has gone away"错误。解决方案：

# 方法1：设置自动重连参数 connection = pymysql.connect( host='localhost', user='user', password='pass', database='test', connect_timeout=10, # 连接超时时间 read_timeout=30, # 读取超时时间 write_timeout=30, # 写入超时时间 ping=1 # 每次执行前ping服务器检查连接 ) # 方法2：使用连接池并设置连接回收时间 pool = PooledDB( creator=pymysql, host='localhost', user='user', password='pass', database='test', ping=1, # 每次取出连接时检查 maxusage=100, # 每个连接最多使用次数 idle_timeout=3600 # 连接空闲超时时间(秒) )

4.2 字符编码问题

处理中文或其他非ASCII字符时，确保正确设置字符集：

# 推荐使用utf8mb4字符集 connection = pymysql.connect( host='localhost', user='user', password='pass', database='test', charset='utf8mb4', # 支持完整的Unicode字符 collation='utf8mb4_unicode_ci' # 排序规则 )

常见编码问题解决方案：

1. 确保数据库、表和字段都使用utf8mb4字符集
2. 连接时明确指定charset='utf8mb4'
3. Python脚本文件本身保存为UTF-8编码
4. 终端或IDE也使用UTF-8编码显示

4.3 性能监控与优化

对于性能敏感的应用，可以监控SQL执行时间：

import time import pymysql def query_with_timing(sql, params=None): connection = pymysql.connect(host='localhost', user='user', password='pass', database='test') try: with connection.cursor() as cursor: start_time = time.time() cursor.execute(sql, params or ()) result = cursor.fetchall() elapsed = time.time() - start_time print(f"SQL执行时间: {elapsed:.3f}秒") return result finally: connection.close()

性能优化建议：

1. 为常用查询条件添加索引
2. 避免SELECT *，只查询需要的字段
3. 合理使用JOIN，避免过度连接
4. 对于复杂查询，考虑使用EXPLAIN分析执行计划
5. 定期优化表（OPTIMIZE TABLE）