1. PyMySQL基础入门与安装配置PyMySQL是Python中操作MySQL数据库最常用的纯Python驱动库之一。作为PEP 249规范的实现它提供了与MySQL服务器交互的标准接口。相比MySQL官方提供的MySQL Connector/PythonPyMySQL具有纯Python实现的优势无需编译即可使用特别适合快速开发和跨平台部署。1.1 环境准备与安装PyMySQL对运行环境有明确要求Python版本CPython 3.9 或 PyPy最新3.x版本MySQL服务器MySQL LTS版本或MariaDB LTS版本安装基础包只需执行pip install pymysql对于需要更高级认证方式的场景可以安装额外依赖# RSA加密认证支持 pip install pymysql[rsa] # MariaDB的ed25519认证支持 pip install pymysql[ed25519]注意在生产环境中建议固定PyMySQL的版本号以避免意外升级带来的兼容性问题例如使用pip install pymysql1.0.2。1.2 基本连接配置建立数据库连接的基本参数包括import pymysql conn pymysql.connect( hostlocalhost, # 数据库服务器地址 userusername, # 数据库用户名 passwordpwd, # 数据库密码 databasedbname, # 默认数据库名 port3306, # 端口默认为3306 charsetutf8mb4, # 字符集 cursorclasspymysql.cursors.DictCursor # 返回字典格式的结果 )关键参数说明charset强烈建议设置为utf8mb4以支持完整的Unicode字符包括emojicursorclass决定了查询结果的返回形式DictCursor会返回字典而非元组连接默认不自动提交事务需要手动调用conn.commit()2. 核心CRUD操作详解2.1 数据插入操作插入数据的基本模式with conn.cursor() as cursor: sql INSERT INTO users (name, email) VALUES (%s, %s) cursor.execute(sql, (张三, zhangsanexample.com)) conn.commit() # 提交事务批量插入的高效写法data [(李四, lisitest.com), (王五, wangwutest.com)] with conn.cursor() as cursor: cursor.executemany( INSERT INTO users (name, email) VALUES (%s, %s), data ) conn.commit()经验大批量插入时使用executemany比循环调用execute效率高得多实测10万条数据插入时间可从分钟级降到秒级。2.2 数据查询与结果处理基本查询示例with conn.cursor() as cursor: cursor.execute(SELECT id, name FROM users WHERE id %s, (5,)) results cursor.fetchall() for row in results: print(fID: {row[id]}, Name: {row[name]})结果获取方法对比方法返回适用场景fetchone()单条记录只需要第一条结果时fetchmany(size)指定数量的记录分批处理大数据集fetchall()所有记录结果集较小时2.3 更新与删除操作更新数据示例with conn.cursor() as cursor: cursor.execute( UPDATE users SET email%s WHERE id%s, (new_emailtest.com, 10) ) conn.commit()删除数据注意事项try: with conn.cursor() as cursor: affected_rows cursor.execute( DELETE FROM users WHERE last_login %s, (2023-01-01,) ) conn.commit() print(f删除了{affected_rows}条记录) except Exception as e: conn.rollback() # 出错时回滚 print(f删除失败: {e})3. 高级特性与性能优化3.1 事务管理与隔离级别PyMySQL支持标准的事务操作try: conn.begin() # 显式开始事务 with conn.cursor() as cursor: cursor.execute(UPDATE account SET balancebalance-100 WHERE user_id1) cursor.execute(UPDATE account SET balancebalance100 WHERE user_id2) conn.commit() except: conn.rollback()设置隔离级别from pymysql.constants import CLIENT conn pymysql.connect( # ...其他参数... client_flagCLIENT.MULTI_STATEMENTS, isolation_levelREPEATABLE READ # 可选的隔离级别 )3.2 连接池管理对于高并发应用建议使用连接池from pymysql import Pool # 创建连接池 pool Pool( min2, # 最小连接数 max10, # 最大连接数 hostlocalhost, userroot, passwordpassword, databasetest ) # 使用连接 conn pool.get_conn() try: with conn.cursor() as cursor: cursor.execute(SELECT * FROM users) print(cursor.fetchall()) finally: pool.release(conn) # 释放连接回池3.3 预处理语句与SQL注入防护PyMySQL使用参数化查询来防止SQL注入# 安全的方式 - 使用参数化查询 cursor.execute(SELECT * FROM users WHERE id %s, (user_id,)) # 危险的方式 - 字符串拼接绝对避免 cursor.execute(fSELECT * FROM users WHERE id {user_id}) # SQL注入风险4. 实战技巧与常见问题4.1 数据类型映射与处理常见数据类型处理建议MySQL类型Python类型处理建议DATETIMEdatetime使用conn.converter配置转换DECIMALDecimal确保导入decimal模块BLOBbytes直接读写二进制数据处理日期时间的正确方式from datetime import datetime # 写入日期 cursor.execute( INSERT INTO posts (title, created_at) VALUES (%s, %s), (标题, datetime.now()) ) # 读取日期 cursor.execute(SELECT created_at FROM posts WHERE id1) row cursor.fetchone() print(type(row[created_at])) # 应该得到datetime类型4.2 性能优化实践批量操作使用executemany替代循环单条插入服务器端游标大数据集时使用SSCursorcursor conn.cursor(pymysql.cursors.SSCursor)合理设置prepared参数高频重复查询可设置为True连接复用避免频繁创建销毁连接4.3 错误处理与调试常见错误及解决方案错误类型可能原因解决方案OperationalError连接问题检查网络、权限、服务状态ProgrammingErrorSQL语法错误检查SQL语句使用参数化查询IntegrityError约束冲突检查唯一性约束、外键约束推荐的错误处理模式try: with conn.cursor() as cursor: cursor.execute(SELECT * FROM non_existent_table) except pymysql.MySQLError as e: print(fMySQL错误 {e.args[0]}: {e.args[1]}) if e.args[0] 1146: # 表不存在错误码 print(请检查表名是否正确) finally: conn.close()4.4 与框架集成在Web框架中的典型用法以Flask为例import pymysql from flask import Flask, g app Flask(__name__) def get_db(): if db not in g: g.db pymysql.connect( hostlocalhost, userflask_user, passwordpassword, databaseflask_db ) return g.db app.teardown_appcontext def close_db(eNone): db g.pop(db, None) if db is not None: db.close() app.route(/users) def list_users(): db get_db() with db.cursor() as cursor: cursor.execute(SELECT id, username FROM users) users cursor.fetchall() return {users: users}在Django中的配置作为备用数据库DATABASES { default: { ENGINE: django.db.backends.mysql, OPTIONS: { read_default_file: /path/to/my.cnf, }, }, legacy: { ENGINE: django.db.backends.mysql, NAME: legacy_db, USER: legacy_user, PASSWORD: password, HOST: legacy.example.com, PORT: 3306, OPTIONS: { init_command: SET sql_modeSTRICT_TRANS_TABLES, }, } }在实际项目中使用PyMySQL时我发现连接管理是最容易出问题的环节。一个实用的建议是使用上下文管理器封装连接操作这样可以确保连接总是被正确关闭from contextlib import contextmanager contextmanager def db_connection(): conn pymysql.connect( hostlocalhost, useruser, passwordpwd, databasedb ) try: yield conn finally: conn.close() # 使用示例 with db_connection() as conn: with conn.cursor() as cursor: cursor.execute(SELECT * FROM products) print(cursor.fetchall())对于需要处理大量数据的场景推荐使用服务器端游标(SSCursor)并配合生成器逐行处理这样可以显著降低内存消耗def iterate_large_table(table_name, batch_size1000): with db_connection() as conn: with conn.cursor(pymysql.cursors.SSCursor) as cursor: cursor.execute(fSELECT * FROM {table_name}) while True: rows cursor.fetchmany(batch_size) if not rows: break for row in rows: yield row