当前位置： 首页 > news >正文

【Python】PyMySQL 实战：从基础连接到高效数据操作

news 2026/7/14 11:18:24

1. PyMySQL快速入门：连接MySQL数据库

第一次用PyMySQL操作数据库时，我踩了不少坑。记得当时连最简单的连接语句都写错，把port参数写成字符串导致报错。现在回头看，这些经验反而成了最好的教学案例。

PyMySQL是Python3中最流行的MySQL驱动库，它用纯Python实现，安装简单到只需一行命令：

pip install pymysql

安装完成后，我们可以用5行代码建立数据库连接：

import pymysql conn = pymysql.connect( host='localhost', # 数据库IP，本地可以用127.0.0.1 user='your_username', # 你的MySQL用户名 password='your_password', # 你的MySQL密码 database='test_db', # 要连接的数据库名 port=3306, # MySQL默认端口 charset='utf8mb4' # 推荐使用utf8mb4支持emoji )

这里有几个新手常犯的错误：

  1. 忘记指定charset参数，导致中文乱码
  2. port参数应该传整数而非字符串
  3. 密码包含特殊字符时没加引号

连接成功后，我们需要创建游标对象来执行SQL语句：

cursor = conn.cursor() # 创建普通游标 dict_cursor = conn.cursor(pymysql.cursors.DictCursor) # 创建返回字典的游标

实际项目中，我强烈推荐使用DictCursor，它会让返回结果以字典形式呈现，通过字段名而非数字索引访问数据，代码可读性大幅提升。

验证连接是否成功的最佳方式是执行一个简单查询：

cursor.execute("SELECT VERSION()") version = cursor.fetchone() print(f"MySQL版本: {version[0]}")

最后别忘了关闭连接，这是很多初学者会遗漏的重要步骤：

cursor.close() # 先关闭游标 conn.close() # 再关闭连接

我见过太多因为未正确关闭连接导致数据库连接数耗尽的生产事故。在后续章节我们会介绍更安全的上下文管理器用法。

2. CRUD操作实战：增删改查全掌握

数据库操作的核心就是CRUD（Create, Read, Update, Delete）。让我用一个用户管理系统的例子，带你完整走一遍这四种操作。

2.1 创建数据表

首先我们创建用户表，包含id、用户名、邮箱和注册时间字段：

create_table_sql = """ CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(50) NOT NULL UNIQUE, email VARCHAR(100) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ cursor.execute(create_table_sql) conn.commit() # DDL语句也需要commit

这里有几个设计要点：

  • 使用AUTO_INCREMENT让ID自增
  • UNIQUE约束保证用户名唯一
  • 使用InnoDB引擎支持事务
  • DEFAULT CURRENT_TIMESTAMP自动设置创建时间

2.2 插入数据

插入单条数据的标准写法：

insert_sql = "INSERT INTO users(username, email) VALUES(%s, %s)" try: cursor.execute(insert_sql, ('python_lover', 'python@example.com')) conn.commit() print(f"新增用户ID: {cursor.lastrowid}") # 获取自增ID except pymysql.IntegrityError: print("用户名已存在") conn.rollback()

批量插入数据时，使用executemany效率更高：

users = [ ('user1', 'user1@test.com'), ('user2', 'user2@test.com'), ('user3', 'user3@test.com') ] cursor.executemany(insert_sql, users) conn.commit()

2.3 查询数据

基础查询非常简单：

cursor.execute("SELECT id, username, email FROM users") for row in cursor.fetchall(): print(f"ID: {row[0]}, 用户名: {row[1]}, 邮箱: {row[2]}")

使用字典游标会让代码更清晰：

dict_cursor.execute("SELECT * FROM users WHERE username=%s", ('python_lover',)) user = dict_cursor.fetchone() print(f"用户邮箱: {user['email']}")

分页查询是常见需求：

page = 1 page_size = 10 offset = (page - 1) * page_size cursor.execute( "SELECT * FROM users ORDER BY id DESC LIMIT %s OFFSET %s", (page_size, offset) )

2.4 更新与删除

更新用户信息：

update_sql = "UPDATE users SET email=%s WHERE username=%s" try: cursor.execute(update_sql, ('new_email@test.com', 'python_lover')) conn.commit() print(f"影响行数: {cursor.rowcount}") except: conn.rollback()

删除用户：

delete_sql = "DELETE FROM users WHERE username=%s" cursor.execute(delete_sql, ('inactive_user',)) conn.commit()

3. 高级技巧：提升操作效率与安全性

当数据量增大后，基础CRUD操作的性能问题就会显现。下面分享几个我在实战中总结的优化技巧。

3.1 批量操作提速

executemany的底层其实是循环执行单条SQL，对于大批量插入，我们可以使用更高效的方式：

# 传统方式 - 慢 data = [(f'user_{i}', f'user_{i}@test.com') for i in range(1000)] cursor.executemany(insert_sql, data) # 优化方式 - 快 values = ','.join([cursor.mogrify('(%s,%s)', row).decode('utf-8') for row in data]) cursor.execute(f"INSERT INTO users(username,email) VALUES {values}")

实测万条数据插入时间从12秒降到0.8秒。原理是合并多条INSERT为一条SQL，减少网络往返。

3.2 事务处理

转账场景是经典的事务案例：

def transfer_money(conn, from_user, to_user, amount): try: with conn.cursor() as cursor: # 检查余额 cursor.execute("SELECT balance FROM accounts WHERE user=%s", (from_user,)) balance = cursor.fetchone()[0] if balance < amount: raise ValueError("余额不足") # 扣款 cursor.execute( "UPDATE accounts SET balance=balance-%s WHERE user=%s", (amount, from_user) ) # 入账 cursor.execute( "UPDATE accounts SET balance=balance+%s WHERE user=%s", (amount, to_user) ) conn.commit() # 提交事务 except: conn.rollback() # 回滚事务 raise

关键点：

  • 使用with语句确保游标自动关闭
  • 所有操作成功后才commit
  • 任何异常触发rollback

3.3 防SQL注入

永远不要拼接SQL字符串：

# 危险！可能被SQL注入 username = input("用户名: ") cursor.execute(f"SELECT * FROM users WHERE username='{username}'") # 安全方式 cursor.execute("SELECT * FROM users WHERE username=%s", (username,))

PyMySQL的参数化查询会自动处理特殊字符转义，这是最基本的安全防护。

4. 上下文管理器与连接池

4.1 使用with自动管理资源

手动管理连接容易遗忘关闭操作，Python的上下文管理器是更好的选择：

class DB: def __init__(self, **kwargs): self.conn_kwargs = kwargs def __enter__(self): self.conn = pymysql.connect(**self.conn_kwargs) return self.conn.cursor(pymysql.cursors.DictCursor) def __exit__(self, exc_type, exc_val, exc_tb): self.conn.commit() if not exc_type else self.conn.rollback() self.conn.close() # 使用示例 with DB(host='localhost', user='root', database='test') as cursor: cursor.execute("SELECT * FROM users") for row in cursor: print(row)

这种写法确保在任何情况下连接都会被正确关闭，事务也会根据是否发生异常自动提交或回滚。

4.2 连接池优化性能

Web应用中频繁创建关闭连接会消耗大量资源，连接池是必备优化：

from dbutils.pooled_db import PooledDB pool = PooledDB( creator=pymysql, maxconnections=20, # 最大连接数 host='localhost', user='root', database='test' ) def get_users(): conn = pool.connection() try: with conn.cursor() as cursor: cursor.execute("SELECT * FROM users") return cursor.fetchall() finally: conn.close()

连接池的主要参数：

  • mincached: 初始空闲连接数
  • maxcached: 最大空闲连接数
  • maxshared: 最大共享连接数
  • blocking: 当连接不足时是否阻塞等待

5. 实战：用户管理系统

综合运用前面所学，我们实现一个完整的用户管理系统：

import pymysql from datetime import datetime class UserManager: def __init__(self): self.conn = pymysql.connect( host='localhost', user='root', database='user_system', cursorclass=pymysql.cursors.DictCursor ) def add_user(self, username, email): with self.conn.cursor() as cursor: try: cursor.execute( "INSERT INTO users(username, email) VALUES(%s, %s)", (username, email) ) self.conn.commit() return cursor.lastrowid except pymysql.IntegrityError: self.conn.rollback() raise ValueError("用户名已存在") def get_user(self, user_id): with self.conn.cursor() as cursor: cursor.execute( "SELECT id, username, email, created_at FROM users WHERE id=%s", (user_id,) ) return cursor.fetchone() def list_users(self, page=1, per_page=10): with self.conn.cursor() as cursor: offset = (page - 1) * per_page cursor.execute( "SELECT * FROM users ORDER BY id DESC LIMIT %s OFFSET %s", (per_page, offset) ) return cursor.fetchall() def update_email(self, user_id, new_email): with self.conn.cursor() as cursor: cursor.execute( "UPDATE users SET email=%s WHERE id=%s", (new_email, user_id) ) self.conn.commit() return cursor.rowcount def delete_user(self, user_id): with self.conn.cursor() as cursor: cursor.execute("DELETE FROM users WHERE id=%s", (user_id,)) self.conn.commit() return cursor.rowcount def __del__(self): self.conn.close() # 使用示例 if __name__ == '__main__': manager = UserManager() # 添加用户 user_id = manager.add_user('test_user', 'test@example.com') # 查询用户 user = manager.get_user(user_id) print(f"新用户: {user['username']}") # 分页列表 print("用户列表:") for user in manager.list_users(): print(f"{user['id']}: {user['username']}") # 更新邮箱 manager.update_email(user_id, 'new_email@test.com') # 删除用户 manager.delete_user(user_id)

这个案例涵盖了PyMySQL的大部分核心功能，包括：

  • 连接管理
  • 事务处理
  • 参数化查询
  • 分页查询
  • 异常处理

在实际开发中，你可以基于这个模板扩展更多功能，如用户登录验证、密码加密、权限管理等。

http://www.jsqmd.com/news/1186732/

相关文章：

  • 电力系统潮流估算基础：从电压降落与功率损耗到线路运行特性分析
  • 2026南京市民问：回收黄金要不要票据？正规流程一文看懂 - 禹竞奢收行
  • OBS多平台直播插件深度指南：从入门到精通的5大进阶技巧
  • 从零到一：构建你的大语言模型面试知识体系
  • 2026年7月最新海口伯爵官方售后服务网点地址及客服电话一览 - 亨得利钟表维修中心
  • 多维聚合实战指南：从Pandas groupby到生产级风控分析
  • SCP收容物041~050：从心灵瘟疫到被诅咒编号的异常档案
  • 2026专业投标书制作服务商甄选全攻略：行业趋势、选型标准、优质机构盘点及避坑FAQ - 产业观察报
  • TSSE-driver在企业环境中的部署策略与最佳实践
  • 2026大连口碑好的漏水检测推荐汇总 - 谁都没有我好看
  • LosslessCut终极指南：无损视频剪辑的快速解决方案
  • Vue 3 实战：从零到一构建电商项目小兔鲜
  • 【电子元器件篇】2.电容：从选型到实战，深入解析高频电路中的去耦与旁路设计
  • 告别乱码：Matplotlib与Seaborn图表中文显示的实战避坑指南
  • 视频异常检测 综述（二）：方法演进与评估新范式
  • Zero-Shot Noise2Noise：轻量级网络如何实现单图自监督高效去噪
  • Unity协程中yield break的五大误用场景与解决方案
  • 2026厦门包包回收避坑攻略：正规门店实测高价变现指南 - 奢侈品回收机构参考
  • 从零开始掌握KLayout Python集成：芯片设计自动化的完整指南
  • 解构Anko：Kotlin时代Android开发的声明式UI革命
  • nomic-embed-text-v1.5边缘部署实战：从4GB到1GB的内存优化方案
  • 从 SpringAI Alibaba 到 RAG + Milvus：未来传统软件会走向何方？
  • 【深度测评】厦门黄金回收｜湖里区五缘湾商圈金条足金分开计价不打包混价克扣总价 - 讯息早知道
  • ADP5350与PIC24FJ1024GB610在嵌入式电源管理中的实战应用
  • 【python零基础教程第22讲】Python性能优化实战
  • gala-anteater核心功能揭秘：一文读懂时序异常检测平台的底层架构
  • MSPM0G350x嵌入式开发实战：PMU、时钟、DMA与事件管理器协同设计
  • 大连奢侈品上门回收实测｜名表黄金包包全收，正规资质不踩坑 - 融媒生活
  • Pandas六大高危操作：生产环境中的数据准确性守门员
  • Apache Sedona架构解析：分布式地理空间计算的工程实践与性能优化