1. Python3与MySQL数据库交互基础
PyMySQL是Python3中用于连接MySQL数据库的纯Python实现库，它完全遵循Python DB API 2.0规范。与MySQLdb相比，PyMySQL不需要编译安装，兼容性更好，特别适合Python3环境。
1.1 环境准备与安装
在开始使用PyMySQL之前，需要确保已安装Python3和MySQL数据库。推荐使用Python 3.6+和MySQL 5.7+/8.0版本组合，这是目前最稳定的搭配。
安装PyMySQL非常简单，使用pip命令即可：
pip install PyMySQL对于需要特定版本的情况，可以指定版本号：
pip install PyMySQL==1.0.2注意：如果同时安装了MySQLdb和PyMySQL，建议优先使用PyMySQL，因为它在Python3中的支持更好，且维护更活跃。
1.2 基本连接配置
建立数据库连接是操作MySQL的第一步，以下是基本连接示例：
import pymysql # 建立数据库连接 connection = pymysql.connect( host='localhost', # 数据库服务器地址 user='username', # 数据库用户名 password='password', # 数据库密码 database='test_db', # 数据库名 port=3306, # MySQL默认端口 charset='utf8mb4', # 字符编码 cursorclass=pymysql.cursors.DictCursor # 设置返回字典格式的结果 ) try: with connection.cursor() as cursor: # 执行SQL查询 sql = "SELECT * FROM users WHERE id = %s" cursor.execute(sql, (1,)) # 获取查询结果 result = cursor.fetchone() print(result) finally: # 关闭连接 connection.close()连接参数说明：
host: MySQL服务器地址，本地可以使用'localhost'或'127.0.0.1'user: 数据库用户名password: 对应用户的密码database: 要连接的数据库名称port: MySQL服务端口，默认3306charset: 字符集编码，推荐使用'utf8mb4'以支持完整的Unicode字符cursorclass: 设置游标类型，DictCursor会返回字典形式的结果
2. 数据库基本操作详解
2.1 创建表操作
使用PyMySQL执行DDL语句创建表：
def create_table(): connection = pymysql.connect(host='localhost', user='user', password='passwd', database='test_db', charset='utf8mb4') try: with connection.cursor() as cursor: # 创建users表 sql = """ CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(50) NOT NULL UNIQUE, email VARCHAR(100) NOT NULL UNIQUE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ cursor.execute(sql) # 提交事务 connection.commit() finally: connection.close()表设计注意事项：
- 主键通常使用自增INT类型
- 字符串字段明确指定字符集和排序规则
- 重要的字段添加NOT NULL约束
- 唯一性字段添加UNIQUE约束
- 时间戳字段设置合适的默认值
2.2 数据插入操作
插入数据是数据库操作的基础，PyMySQL提供了多种插入方式：
def insert_data(): connection = pymysql.connect(...) # 连接参数同上 try: with connection.cursor() as cursor: # 单条插入 sql = "INSERT INTO users (username, email) VALUES (%s, %s)" cursor.execute(sql, ('user1', 'user1@example.com')) # 批量插入 users = [ ('user2', 'user2@example.com'), ('user3', 'user3@example.com'), ('user4', 'user4@example.com') ] cursor.executemany(sql, users) connection.commit() except pymysql.err.IntegrityError as e: print(f"插入数据失败: {e}") connection.rollback() finally: connection.close()插入数据时的最佳实践：
- 始终使用参数化查询(%s占位符)而非字符串拼接，防止SQL注入
- 批量操作使用executemany()提高效率
- 处理可能的异常，如唯一键冲突(IntegrityError)
- 操作完成后及时提交或回滚事务
2.3 数据查询操作
PyMySQL提供了多种数据查询和结果获取方式：
def query_data(): connection = pymysql.connect(...) try: with connection.cursor() as cursor: # 基本查询 cursor.execute("SELECT * FROM users WHERE username LIKE %s", ('user%',)) # 获取所有结果 all_results = cursor.fetchall() print("所有结果:", all_results) # 获取单条结果 cursor.execute("SELECT * FROM users WHERE id = %s", (1,)) one_result = cursor.fetchone() print("单条结果:", one_result) # 分批获取结果 cursor.execute("SELECT * FROM users") while True: batch = cursor.fetchmany(size=2) # 每次获取2条 if not batch: break print("批次结果:", batch) finally: connection.close()查询结果处理技巧：
- fetchall()返回所有结果，适合小数据量
- fetchone()获取单条结果，常用于精确查询
- fetchmany(size)分批获取，适合大数据量处理
- 游标会保持状态，可以多次执行不同查询
3. 高级功能与性能优化
3.1 事务处理
MySQL的事务特性对于数据一致性至关重要：
def transfer_money(from_id, to_id, amount): connection = pymysql.connect(...) try: with connection.cursor() as cursor: # 检查转出账户余额 cursor.execute("SELECT balance FROM accounts WHERE id = %s FOR UPDATE", (from_id,)) from_balance = cursor.fetchone()['balance'] if from_balance < amount: raise ValueError("余额不足") # 扣减转出账户 cursor.execute("UPDATE accounts SET balance = balance - %s WHERE id = %s", (amount, from_id)) # 增加转入账户 cursor.execute("UPDATE accounts SET balance = balance + %s WHERE id = %s", (amount, to_id)) connection.commit() except Exception as e: connection.rollback() print(f"转账失败: {e}") finally: connection.close()事务处理要点：
- 使用FOR UPDATE锁定要修改的行，防止并发修改
- 在try块中执行所有数据库操作
- 成功时提交(commit)，失败时回滚(rollback)
- 确保在finally中关闭连接
3.2 连接池管理
对于高并发应用，使用连接池可以显著提高性能：
from dbutils.pooled_db import PooledDB # 创建连接池 pool = PooledDB( creator=pymysql, maxconnections=10, # 最大连接数 mincached=2, # 初始化时创建的连接数 host='localhost', user='user', password='passwd', database='test_db', charset='utf8mb4' ) def query_with_pool(): # 从连接池获取连接 connection = pool.connection() try: with connection.cursor() as cursor: cursor.execute("SELECT * FROM users") results = cursor.fetchall() return results finally: # 将连接返回连接池而非关闭 connection.close()连接池配置建议：
- maxconnections根据应用负载调整，通常10-50
- mincached设置初始连接数，减少首次请求延迟
- 使用后调用connection.close()将连接返回到池中
- 考虑使用连接池管理工具如SQLAlchemy
3.3 预处理语句与性能
预处理语句可以提高性能并防止SQL注入：
def prepared_statement(): connection = pymysql.connect(...) try: with connection.cursor() as cursor: # 创建预处理语句 stmt = "INSERT INTO logs (user_id, action) VALUES (%s, %s)" # 批量执行 actions = [ (1, 'login'), (2, 'view_page'), (3, 'logout') ] cursor.executemany(stmt, actions) connection.commit() finally: connection.close()性能优化技巧：
- 对于重复执行的SQL，使用预处理语句
- 批量操作使用executemany()
- 合理使用索引提高查询效率
- 考虑使用存储过程处理复杂逻辑
4. 常见问题与解决方案
4.1 连接问题排查
常见连接错误及解决方法：
错误2003 (HY000): Can't connect to MySQL server
- 检查MySQL服务是否运行
- 确认连接参数(host, port)正确
- 检查防火墙设置
错误1045 (28000): Access denied
- 确认用户名密码正确
- 检查用户是否有远程连接权限
- MySQL8.0+可能需要使用新的认证插件
错误2013 (HY000): Lost connection
- 增加连接超时时间：
connect_timeout=10 - 检查网络稳定性
- 可能是服务器端超时设置过短
- 增加连接超时时间：
连接参数调整示例：
connection = pymysql.connect( host='localhost', user='user', password='passwd', database='test_db', connect_timeout=10, # 连接超时时间(秒) read_timeout=30, # 读取超时时间 write_timeout=30 # 写入超时时间 )4.2 字符编码问题
MySQL字符集常见问题处理：
乱码问题
- 确保连接指定charset='utf8mb4'
- 检查表/字段字符集设置
- Python3字符串处理使用unicode
emoji存储问题
- 必须使用utf8mb4字符集
- 修改表字段定义：
ALTER TABLE messages MODIFY content TEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
排序规则问题
- 中文排序使用utf8mb4_unicode_ci
- 区分大小写使用utf8mb4_bin
4.3 数据类型映射
Python与MySQL数据类型对应关系：
| Python类型 | MySQL类型 | 注意事项 |
|---|---|---|
| int | INT, BIGINT | 注意整数范围 |
| float | FLOAT, DOUBLE | 精度问题需注意 |
| str | VARCHAR, TEXT | 长度限制和字符集很重要 |
| bytes | BLOB, BINARY | 适合存储二进制数据 |
| datetime.datetime | DATETIME, TIMESTAMP | 时区处理要小心 |
| bool | TINYINT(1) | MySQL没有真正的布尔类型 |
类型处理示例：
def type_handling(): connection = pymysql.connect(...) try: with connection.cursor() as cursor: # 处理各种数据类型 data = { 'name': '张三', # str -> VARCHAR 'age': 30, # int -> INT 'score': 89.5, # float -> FLOAT 'is_active': True, # bool -> TINYINT(1) 'birthday': datetime.date(1990, 5, 15), # date -> DATE 'created_at': datetime.datetime.now() # datetime -> DATETIME } sql = """INSERT INTO people (name, age, score, is_active, birthday, created_at) VALUES (%(name)s, %(age)s, %(score)s, %(is_active)s, %(birthday)s, %(created_at)s)""" cursor.execute(sql, data) connection.commit() finally: connection.close()4.4 连接池最佳实践
生产环境连接池配置建议：
大小设置
- 连接池大小 = (核心数 * 2) + 有效磁盘数
- 通常8-50之间，根据负载测试调整
连接验证
- 设置ping=1自动验证连接有效性
- 配置连接最大存活时间
完整配置示例
pool = PooledDB( creator=pymysql, maxconnections=20, mincached=5, maxcached=10, maxusage=100, # 单个连接最大使用次数 blocking=True, # 达到最大连接数时阻塞而非报错 host='localhost', user='user', password='passwd', database='test_db', charset='utf8mb4', ping=1 # 每次使用前ping服务器检查连接 )在实际项目中，PyMySQL与MySQL的交互远不止基本的CRUD操作。掌握连接管理、事务处理、性能优化等高级特性，才能构建健壮的数据库应用。根据具体场景合理选择方案，比如简单应用直接使用PyMySQL，复杂应用可以考虑集成SQLAlchemy等ORM工具。
