尧图网站建设 
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( hostlocalhost, # 数据库服务器地址 userusername, # 数据库用户名 passwordpassword, # 数据库密码 databasedbname, # 数据库名 port3306, # MySQL默认端口 charsetutf8mb4, # 字符集 cursorclasspymysql.cursors.DictCursor # 返回字典形式的结果 )连接参数说明host: MySQL服务器地址可以是IP或域名user: 数据库用户名password: 用户密码database: 要连接的数据库名port: MySQL服务端口默认3306charset: 字符集推荐使用utf8mb4以支持完整的Unicode字符cursorclass: 游标类型DictCursor会返回字典形式的结果1.3 连接池管理在高并发应用中频繁创建和关闭连接会影响性能。PyMySQL本身不提供连接池功能但可以通过第三方库如DBUtils来实现from dbutils.pooled_db import PooledDB import pymysql # 创建连接池 pool PooledDB( creatorpymysql, maxconnections20, # 连接池最大连接数 mincached5, # 初始化时创建的连接数 hostlocalhost, useruser, passwordpass, databasetest, charsetutf8mb4, cursorclasspymysql.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(hostlocalhost, useruser, passwordpass, databasetest) 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) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COLLATEutf8mb4_unicode_ci cursor.execute(sql) connection.commit() finally: connection.close()表设计注意事项使用InnoDB引擎它支持事务和外键字符集使用utf8mb4以支持完整的Unicode字符包括emoji为常用查询字段添加索引为必填字段设置NOT NULL约束为唯一性字段添加UNIQUE约束2.2 插入数据插入数据是最基本的操作之一PyMySQL支持单条插入和批量插入def insert_user(name, email): connection pymysql.connect(hostlocalhost, useruser, passwordpass, databasetest) 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(hostlocalhost, useruser, passwordpass, databasetest) 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(hostlocalhost, useruser, passwordpass, databasetest) 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(hostlocalhost, useruser, passwordpass, databasetest) 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(hostlocalhost, useruser, passwordpass, databasetest) 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(hostlocalhost, useruser, passwordpass, databasetest) 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()事务使用要点明确调用begin()开始事务在try块中执行所有操作成功时调用commit()失败时调用rollback()确保在finally块中关闭连接3.2 批量操作优化对于大量数据的插入或更新批量操作可以显著提高性能def bulk_insert_users(user_data): connection pymysql.connect(hostlocalhost, useruser, passwordpass, databasetest) 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()批量操作优化技巧使用executemany()代替循环执行execute()合理设置批量操作的大小通常1000-5000条记录一批在批量操作中使用事务考虑使用LOAD DATA INFILE对于超大数据量3.3 预处理语句预处理语句可以提高性能并防止SQL注入def get_user_by_id(user_id): connection pymysql.connect(hostlocalhost, useruser, passwordpass, databasetest) 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()预处理语句的优势数据库只需解析SQL一次自动处理参数转义防止SQL注入对于重复执行的查询性能更好4. 常见问题与解决方案4.1 连接超时问题MySQL服务器默认会在8小时不活动后关闭连接这会导致PyMySQL抛出MySQL server has gone away错误。解决方案# 方法1设置自动重连参数 connection pymysql.connect( hostlocalhost, useruser, passwordpass, databasetest, connect_timeout10, # 连接超时时间 read_timeout30, # 读取超时时间 write_timeout30, # 写入超时时间 ping1 # 每次执行前ping服务器检查连接 ) # 方法2使用连接池并设置连接回收时间 pool PooledDB( creatorpymysql, hostlocalhost, useruser, passwordpass, databasetest, ping1, # 每次取出连接时检查 maxusage100, # 每个连接最多使用次数 idle_timeout3600 # 连接空闲超时时间(秒) )4.2 字符编码问题处理中文或其他非ASCII字符时确保正确设置字符集# 推荐使用utf8mb4字符集 connection pymysql.connect( hostlocalhost, useruser, passwordpass, databasetest, charsetutf8mb4, # 支持完整的Unicode字符 collationutf8mb4_unicode_ci # 排序规则 )常见编码问题解决方案确保数据库、表和字段都使用utf8mb4字符集连接时明确指定charsetutf8mb4Python脚本文件本身保存为UTF-8编码终端或IDE也使用UTF-8编码显示4.3 性能监控与优化对于性能敏感的应用可以监控SQL执行时间import time import pymysql def query_with_timing(sql, paramsNone): connection pymysql.connect(hostlocalhost, useruser, passwordpass, databasetest) try: with connection.cursor() as cursor: start_time time.time() cursor.execute(sql, params or ()) result cursor.fetchall() elapsed time.time() - start_time print(fSQL执行时间: {elapsed:.3f}秒) return result finally: connection.close()性能优化建议为常用查询条件添加索引避免SELECT *只查询需要的字段合理使用JOIN避免过度连接对于复杂查询考虑使用EXPLAIN分析执行计划定期优化表OPTIMIZE TABLE 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( hostlocalhost, # 数据库服务器地址 userusername, # 数据库用户名 passwordpassword, # 数据库密码 databasedbname, # 数据库名 port3306, # MySQL默认端口 charsetutf8mb4, # 字符集 cursorclasspymysql.cursors.DictCursor # 返回字典形式的结果 )连接参数说明host: MySQL服务器地址可以是IP或域名user: 数据库用户名password: 用户密码database: 要连接的数据库名port: MySQL服务端口默认3306charset: 字符集推荐使用utf8mb4以支持完整的Unicode字符cursorclass: 游标类型DictCursor会返回字典形式的结果1.3 连接池管理在高并发应用中频繁创建和关闭连接会影响性能。PyMySQL本身不提供连接池功能但可以通过第三方库如DBUtils来实现from dbutils.pooled_db import PooledDB import pymysql # 创建连接池 pool PooledDB( creatorpymysql, maxconnections20, # 连接池最大连接数 mincached5, # 初始化时创建的连接数 hostlocalhost, useruser, passwordpass, databasetest, charsetutf8mb4, cursorclasspymysql.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(hostlocalhost, useruser, passwordpass, databasetest) 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) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COLLATEutf8mb4_unicode_ci cursor.execute(sql) connection.commit() finally: connection.close()表设计注意事项使用InnoDB引擎它支持事务和外键字符集使用utf8mb4以支持完整的Unicode字符包括emoji为常用查询字段添加索引为必填字段设置NOT NULL约束为唯一性字段添加UNIQUE约束2.2 插入数据插入数据是最基本的操作之一PyMySQL支持单条插入和批量插入def insert_user(name, email): connection pymysql.connect(hostlocalhost, useruser, passwordpass, databasetest) 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(hostlocalhost, useruser, passwordpass, databasetest) 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(hostlocalhost, useruser, passwordpass, databasetest) 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(hostlocalhost, useruser, passwordpass, databasetest) 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(hostlocalhost, useruser, passwordpass, databasetest) 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(hostlocalhost, useruser, passwordpass, databasetest) 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()事务使用要点明确调用begin()开始事务在try块中执行所有操作成功时调用commit()失败时调用rollback()确保在finally块中关闭连接3.2 批量操作优化对于大量数据的插入或更新批量操作可以显著提高性能def bulk_insert_users(user_data): connection pymysql.connect(hostlocalhost, useruser, passwordpass, databasetest) 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()批量操作优化技巧使用executemany()代替循环执行execute()合理设置批量操作的大小通常1000-5000条记录一批在批量操作中使用事务考虑使用LOAD DATA INFILE对于超大数据量3.3 预处理语句预处理语句可以提高性能并防止SQL注入def get_user_by_id(user_id): connection pymysql.connect(hostlocalhost, useruser, passwordpass, databasetest) 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()预处理语句的优势数据库只需解析SQL一次自动处理参数转义防止SQL注入对于重复执行的查询性能更好4. 常见问题与解决方案4.1 连接超时问题MySQL服务器默认会在8小时不活动后关闭连接这会导致PyMySQL抛出MySQL server has gone away错误。解决方案# 方法1设置自动重连参数 connection pymysql.connect( hostlocalhost, useruser, passwordpass, databasetest, connect_timeout10, # 连接超时时间 read_timeout30, # 读取超时时间 write_timeout30, # 写入超时时间 ping1 # 每次执行前ping服务器检查连接 ) # 方法2使用连接池并设置连接回收时间 pool PooledDB( creatorpymysql, hostlocalhost, useruser, passwordpass, databasetest, ping1, # 每次取出连接时检查 maxusage100, # 每个连接最多使用次数 idle_timeout3600 # 连接空闲超时时间(秒) )4.2 字符编码问题处理中文或其他非ASCII字符时确保正确设置字符集# 推荐使用utf8mb4字符集 connection pymysql.connect( hostlocalhost, useruser, passwordpass, databasetest, charsetutf8mb4, # 支持完整的Unicode字符 collationutf8mb4_unicode_ci # 排序规则 )常见编码问题解决方案确保数据库、表和字段都使用utf8mb4字符集连接时明确指定charsetutf8mb4Python脚本文件本身保存为UTF-8编码终端或IDE也使用UTF-8编码显示4.3 性能监控与优化对于性能敏感的应用可以监控SQL执行时间import time import pymysql def query_with_timing(sql, paramsNone): connection pymysql.connect(hostlocalhost, useruser, passwordpass, databasetest) try: with connection.cursor() as cursor: start_time time.time() cursor.execute(sql, params or ()) result cursor.fetchall() elapsed time.time() - start_time print(fSQL执行时间: {elapsed:.3f}秒) return result finally: connection.close()性能优化建议为常用查询条件添加索引避免SELECT *只查询需要的字段合理使用JOIN避免过度连接对于复杂查询考虑使用EXPLAIN分析执行计划定期优化表OPTIMIZE TABLE