Python操作MySQL数据库pymysql从入门到进阶文章目录Python操作MySQL数据库pymysql从入门到进阶前言一、安装与建立连接二、创建表与插入数据CREATE/INSERT使用四、更新与删除数据UPDATE/DELETE五、事务控制六、连接池高并发场景的解决方案总结✅ 亮点总结适用场景扩展方向前言在绝大多数后端应用和数据分析项目中数据库是不可或缺的基础设施。MySQL作为最流行的关系型数据库之一与Python的配合使用极为广泛。pymysql是一个纯Python实现的MySQL客户端库——纯Python意味着不需要安装MySQL官方的C客户端库部署简单兼容性好。本文将系统讲解pymysql的连接管理、CRUD操作、事务控制和连接池的高级用法覆盖从开发环境到生产部署的完整知识体系。pymysql vs SQLAlchemypymysql是底层驱动直接执行SQL语句SQLAlchemy是高层ORM将表和行映射为类和对象。初学者容易困惑该用哪个——答案是你两个都需要理解pymysql让你掌握数据库操作的本质参数化查询、事务、连接管理SQLAlchemy让你在复杂项目中保持代码的可维护性。本文专注pymysql打地基。一、安装与建立连接连接MySQL是第一步。两个关键决策点字符集的选择和游标类型的选择直接影响后续的数据处理体验。pipinstallpymysql为什么选择 utf8mb4MySQL的utf8字符集只支持最多3字节的UTF-8字符即BMP字符emoji表情如需要4字节只有utf8mb4才能正确存储。如果你的应用涉及用户昵称、评论等可能包含emoji的文本不使用utf8mb4会导致数据截断或插入失败——这是一个线上事故的常见原因。首先确保MySQL服务已启动然后创建数据库连接importpymysql# 建立连接connectionpymysql.connect(hostlocalhost,port3306,userroot,passwordyour_password,databasetest_db,charsetutf8mb4,# 支持emoji等4字节字符cursorclasspymysql.cursors.DictCursor,# 返回字典格式结果)print(数据库连接成功)print(f服务器版本:{connection.get_server_info()})DictCursor vs 默认Cursor默认游标返回元组你需要按位置索引取值row[0]row[1]。在字段多的时候代码可读性极差。DictCursor返回字典row[name]代码清晰且即使SELECT字段顺序改变也不影响取值。几乎所有生产项目都应使用DictCursor——这是花1分钟配置就能持续获益的选择。使用上下文管理器确保连接安全关闭。在生产环境中忘记关闭数据库连接是导致连接池耗尽的头号元凶withpymysql.connect(hostlocalhost,userroot,passwordyour_password,databasetest_db,charsetutf8mb4,cursorclasspymysql.cursors.DictCursor,)asconn:withconn.cursor()ascursor:cursor.execute(SELECT VERSION())resultcursor.fetchone()print(result)二、创建表与插入数据CREATE/INSERTdefinit_database(conn):withconn.cursor()ascursor:# 创建表create_sql CREATE TABLE IF NOT EXISTS employees ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(50) NOT NULL, department VARCHAR(50), salary DECIMAL(10, 2), hire_date DATE, email VARCHAR(100) UNIQUE ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; cursor.execute(create_sql)conn.commit()print(表创建成功)executemany vs 循环executeexecutemany()一次性将多条记录发给MySQL而循环execute()每条记录都发起一次网络往返。当插入1000条记录时executemany()只需1次网络往返速度可以提高10-100倍。这是数据库操作中最简单但回报最高的优化。autocommit 的行为pymysql默认autocommitFalse这意味着每次写操作后必须显式调用conn.commit()否则数据不会持久化到磁盘。很多初学者发现插入好像成功了但数据库里查不到数据原因就是忘了commit。def insert_employees(conn):with conn.cursor() as cursor:# 单条插入sql “INSERT INTO employees (name, department, salary, hire_date, email) VALUES (%s, %s, %s, %s, %s)”cursor.execute(sql, (“张三”, “技术部”, 15000, “2023-01-15”, “zhangsanexample.com”))conn.commit()# 批量插入使用executemany效率更高 employees [ (李四, 销售部, 12000, 2023-03-20, lisiexample.com), (王五, 技术部, 18000, 2023-02-10, wangwuexample.com), (赵六, 市场部, 13000, 2023-05-08, zhaoliuexample.com), (孙七, 技术部, 20000, 2023-01-01, sunqiexample.com), ] cursor.executemany(sql, employees) conn.commit() print(f成功插入 {len(employees) 1} 条记录)使用with pymysql.connect(host“localhost”, user“root”, password“your_password”,database“test_db”, charset“utf8mb4”) as conn:init_database(conn)insert_employees(conn) **安全提示**始终使用%s占位符进行参数化查询**永远不要使用字符串拼接**来构造SQL这会导致SQL注入漏洞。 ## 三、查询数据SELECT python def query_employees(conn): with conn.cursor() as cursor: # 查询所有记录 cursor.execute(SELECT * FROM employees) results cursor.fetchall() for row in results: print(fID:{row[id]} | {row[name]} | {row[department]} | 薪资:{row[salary]}) # 条件查询 sql SELECT name, salary FROM employees WHERE department %s AND salary %s cursor.execute(sql, (技术部, 15000)) high_salary cursor.fetchall() print(f\n技术部高薪员工: {high_salary}) # 只取一条 cursor.execute(SELECT * FROM employees ORDER BY salary DESC LIMIT 1) top_employee cursor.fetchone() print(f\n最高薪资员工: {top_employee}) # 统计查询 cursor.execute( SELECT department, COUNT(*) as count, AVG(salary) as avg_salary FROM employees GROUP BY department ) stats cursor.fetchall() for stat in stats: print(f{stat[department]}: {stat[count]}人, 平均薪资:{stat[avg_salary]:.2f})四、更新与删除数据UPDATE/DELETEdefupdate_employee(conn,emp_id,new_salary):withconn.cursor()ascursor:sqlUPDATE employees SET salary %s WHERE id %saffectedcursor.execute(sql,(new_salary,emp_id))conn.commit()ifaffected:print(f员工ID{emp_id}的薪资已更新为{new_salary})else:print(f未找到ID{emp_id}的员工)defdelete_employee(conn,emp_id):withconn.cursor()ascursor:sqlDELETE FROM employees WHERE id %saffectedcursor.execute(sql,(emp_id,))conn.commit()print(f已删除{affected}条记录)五、事务控制事务是数据库操作的安全网——确保一组数据库操作要么全部成功要么全部回滚。最经典的例子就是银行转账A账户扣100元B账户加100元这两步必须作为一个原子操作——如果第一步成功第二步失败A的钱凭空消失这就是生产事故。事务通过commit()和rollback()机制防止这种问题。deftransfer_salary(conn,from_id,to_id,amount):模拟薪资转账先扣减再增加保证原子性try:withconn.cursor()ascursor:# 开启事务默认已开启autocommitFalse# 扣减cursor.execute(UPDATE employees SET salary salary - %s WHERE id %s,(amount,from_id))# 模拟异常cursor.execute(UPDATE employees SET salary salary %s WHERE id %s,(amount,to_id))# 提交事务conn.commit()print(f转账成功{from_id}-{to_id}金额:{amount})exceptExceptionase:conn.rollback()# 发生异常回滚所有操作print(f转账失败已回滚:{e})六、连接池高并发场景的解决方案频繁创建和销毁数据库连接开销很大连接池复用已有连接大幅提升性能pipinstalldbutils# 或者使用 SQLAlchemy 的连接池fromdbutils.pooled_dbimportPooledDBimportpymysql# 创建连接池poolPooledDB(creatorpymysql,# 使用的数据库模块maxconnections10,# 连接池最大连接数mincached3,# 初始化时创建的空闲连接数maxcached5,# 最大空闲连接数blockingTrue,# 连接数满时是否等待hostlocalhost,userroot,passwordyour_password,databasetest_db,charsetutf8mb4,cursorclasspymysql.cursors.DictCursor,)# 从连接池获取连接defquery_with_pool():connpool.connection()try:withconn.cursor()ascursor:cursor.execute(SELECT COUNT(*) as total FROM employees)resultcursor.fetchone()print(f员工总数:{result[total]})finally:conn.close()# 归还连接到池而非真正关闭# 多线程并发查询示例importthreadingdefworker(name):connpool.connection()try:withconn.cursor()ascursor:cursor.execute(SELECT SLEEP(0.1), %s as name,(name,))print(f[{name}] 查询完成:{cursor.fetchone()})finally:conn.close()threads[threading.Thread(targetworker,args(f线程-{i},))foriinrange(5)]fortinthreads:t.start()fortinthreads:t.join()print(所有线程执行完毕)总结pymysql的学习路径清晰明确连接管理 - CRUD操作 - 事务控制 - 连接池。本文覆盖了从基础连接到生产级连接池的完整技术栈。核心要点包括始终使用参数化查询%s占位符防止SQL注入使用上下文管理器或try/finally确保连接正确释放涉及多个操作时使用事务保证数据一致性高并发场景下使用连接池大幅提升性能掌握这些能力后你就能将Python程序与MySQL数据库无缝对接构建出功能完善的数据驱动型应用。✅ 亮点总结参数化查询%s占位符是抵御SQL注入攻击的第一道防线必须养成使用习惯上下文管理器with connection.cursor()自动管理游标资源代码更简洁安全事务控制commit/rollback确保多步操作的原子性保证数据一致性DBUtils连接池将连接创建开销减少90%以上是高并发场景的标配方案executemany()批量插入比逐条插入快数倍是处理大批量数据的正确姿势适用场景Web后台数据管理Django/Flask项目中连接MySQL存储用户信息、业务数据和日志数据采集与存储爬虫程序将采集的数据通过pymysql写入数据库持久化报表系统数据源配置定时任务从MySQL查询数据用pandas分析后生成报表扩展方向SQLAlchemy ORM学习用面向对象的方式操作数据库告别手写SQL享受模型驱动的开发体验数据库索引与查询优化掌握EXPLAIN分析执行计划、合理创建索引提升查询性能推荐阅读下一篇《Python机器学习入门sklearn》进入人工智能领域的学习之旅