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 PyMySQL1.0.2注意如果同时安装了MySQLdb和PyMySQL建议优先使用PyMySQL因为它在Python3中的支持更好且维护更活跃。1.2 基本连接配置建立数据库连接是操作MySQL的第一步以下是基本连接示例import pymysql # 建立数据库连接 connection pymysql.connect( hostlocalhost, # 数据库服务器地址 userusername, # 数据库用户名 passwordpassword, # 数据库密码 databasetest_db, # 数据库名 port3306, # MySQL默认端口 charsetutf8mb4, # 字符编码 cursorclasspymysql.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.1user: 数据库用户名password: 对应用户的密码database: 要连接的数据库名称port: MySQL服务端口默认3306charset: 字符集编码推荐使用utf8mb4以支持完整的Unicode字符cursorclass: 设置游标类型DictCursor会返回字典形式的结果2. 数据库基本操作详解2.1 创建表操作使用PyMySQL执行DDL语句创建表def create_table(): connection pymysql.connect(hostlocalhost, useruser, passwordpasswd, databasetest_db, charsetutf8mb4) 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 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COLLATEutf8mb4_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, user1example.com)) # 批量插入 users [ (user2, user2example.com), (user3, user3example.com), (user4, user4example.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(size2) # 每次获取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( creatorpymysql, maxconnections10, # 最大连接数 mincached2, # 初始化时创建的连接数 hostlocalhost, useruser, passwordpasswd, databasetest_db, charsetutf8mb4 ) 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-50mincached设置初始连接数减少首次请求延迟使用后调用connection.close()将连接返回到池中考虑使用连接池管理工具如SQLAlchemy3.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): Cant connect to MySQL server检查MySQL服务是否运行确认连接参数(host, port)正确检查防火墙设置错误1045 (28000): Access denied确认用户名密码正确检查用户是否有远程连接权限MySQL8.0可能需要使用新的认证插件错误2013 (HY000): Lost connection增加连接超时时间connect_timeout10检查网络稳定性可能是服务器端超时设置过短连接参数调整示例connection pymysql.connect( hostlocalhost, useruser, passwordpasswd, databasetest_db, connect_timeout10, # 连接超时时间(秒) read_timeout30, # 读取超时时间 write_timeout30 # 写入超时时间 )4.2 字符编码问题MySQL字符集常见问题处理乱码问题确保连接指定charsetutf8mb4检查表/字段字符集设置Python3字符串处理使用unicodeemoji存储问题必须使用utf8mb4字符集修改表字段定义ALTER TABLE messages MODIFY content TEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;排序规则问题中文排序使用utf8mb4_unicode_ci区分大小写使用utf8mb4_bin4.3 数据类型映射Python与MySQL数据类型对应关系Python类型MySQL类型注意事项intINT, BIGINT注意整数范围floatFLOAT, DOUBLE精度问题需注意strVARCHAR, TEXT长度限制和字符集很重要bytesBLOB, BINARY适合存储二进制数据datetime.datetimeDATETIME, TIMESTAMP时区处理要小心boolTINYINT(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之间根据负载测试调整连接验证设置ping1自动验证连接有效性配置连接最大存活时间完整配置示例pool PooledDB( creatorpymysql, maxconnections20, mincached5, maxcached10, maxusage100, # 单个连接最大使用次数 blockingTrue, # 达到最大连接数时阻塞而非报错 hostlocalhost, useruser, passwordpasswd, databasetest_db, charsetutf8mb4, ping1 # 每次使用前ping服务器检查连接 )在实际项目中PyMySQL与MySQL的交互远不止基本的CRUD操作。掌握连接管理、事务处理、性能优化等高级特性才能构建健壮的数据库应用。根据具体场景合理选择方案比如简单应用直接使用PyMySQL复杂应用可以考虑集成SQLAlchemy等ORM工具。 Python3使用PyMySQL操作MySQL数据库全指南
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 PyMySQL1.0.2注意如果同时安装了MySQLdb和PyMySQL建议优先使用PyMySQL因为它在Python3中的支持更好且维护更活跃。1.2 基本连接配置建立数据库连接是操作MySQL的第一步以下是基本连接示例import pymysql # 建立数据库连接 connection pymysql.connect( hostlocalhost, # 数据库服务器地址 userusername, # 数据库用户名 passwordpassword, # 数据库密码 databasetest_db, # 数据库名 port3306, # MySQL默认端口 charsetutf8mb4, # 字符编码 cursorclasspymysql.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.1user: 数据库用户名password: 对应用户的密码database: 要连接的数据库名称port: MySQL服务端口默认3306charset: 字符集编码推荐使用utf8mb4以支持完整的Unicode字符cursorclass: 设置游标类型DictCursor会返回字典形式的结果2. 数据库基本操作详解2.1 创建表操作使用PyMySQL执行DDL语句创建表def create_table(): connection pymysql.connect(hostlocalhost, useruser, passwordpasswd, databasetest_db, charsetutf8mb4) 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 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COLLATEutf8mb4_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, user1example.com)) # 批量插入 users [ (user2, user2example.com), (user3, user3example.com), (user4, user4example.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(size2) # 每次获取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( creatorpymysql, maxconnections10, # 最大连接数 mincached2, # 初始化时创建的连接数 hostlocalhost, useruser, passwordpasswd, databasetest_db, charsetutf8mb4 ) 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-50mincached设置初始连接数减少首次请求延迟使用后调用connection.close()将连接返回到池中考虑使用连接池管理工具如SQLAlchemy3.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): Cant connect to MySQL server检查MySQL服务是否运行确认连接参数(host, port)正确检查防火墙设置错误1045 (28000): Access denied确认用户名密码正确检查用户是否有远程连接权限MySQL8.0可能需要使用新的认证插件错误2013 (HY000): Lost connection增加连接超时时间connect_timeout10检查网络稳定性可能是服务器端超时设置过短连接参数调整示例connection pymysql.connect( hostlocalhost, useruser, passwordpasswd, databasetest_db, connect_timeout10, # 连接超时时间(秒) read_timeout30, # 读取超时时间 write_timeout30 # 写入超时时间 )4.2 字符编码问题MySQL字符集常见问题处理乱码问题确保连接指定charsetutf8mb4检查表/字段字符集设置Python3字符串处理使用unicodeemoji存储问题必须使用utf8mb4字符集修改表字段定义ALTER TABLE messages MODIFY content TEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;排序规则问题中文排序使用utf8mb4_unicode_ci区分大小写使用utf8mb4_bin4.3 数据类型映射Python与MySQL数据类型对应关系Python类型MySQL类型注意事项intINT, BIGINT注意整数范围floatFLOAT, DOUBLE精度问题需注意strVARCHAR, TEXT长度限制和字符集很重要bytesBLOB, BINARY适合存储二进制数据datetime.datetimeDATETIME, TIMESTAMP时区处理要小心boolTINYINT(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之间根据负载测试调整连接验证设置ping1自动验证连接有效性配置连接最大存活时间完整配置示例pool PooledDB( creatorpymysql, maxconnections20, mincached5, maxcached10, maxusage100, # 单个连接最大使用次数 blockingTrue, # 达到最大连接数时阻塞而非报错 hostlocalhost, useruser, passwordpasswd, databasetest_db, charsetutf8mb4, ping1 # 每次使用前ping服务器检查连接 )在实际项目中PyMySQL与MySQL的交互远不止基本的CRUD操作。掌握连接管理、事务处理、性能优化等高级特性才能构建健壮的数据库应用。根据具体场景合理选择方案比如简单应用直接使用PyMySQL复杂应用可以考虑集成SQLAlchemy等ORM工具。