Python MySQL从零上手：30分钟搞定pymysql基本CRUD操作

Python MySQL从零上手：30分钟搞定pymysql基本CRUD操作

刚开始用Python操作MySQL时，我也被各种问题搞得焦头烂额——连接突然断开、SQL注入漏洞、查询结果乱码、事务不知道怎么用。但说实话，一旦掌握了pymysql这个基础工具的正确用法，你会发现数据库操作其实很简单。今天我就带你从零开始，用30分钟搞定MySQL的基本增删改查（CRUD），让你在Python项目中能自信地操作数据库。

一、环境准备：搭建你的第一个Python+MySQL开发环境

1.1 为什么需要pymysql？

你可能会有疑问：Python不是自带数据库模块吗？为什么还要装pymysql？这里有个关键点要理解：

Python本身不直接支持MySQL，它需要通过一个"翻译官"（驱动）来和MySQL对话。pymysql就是这样一个纯Python写的翻译官，它把Python的指令翻译成MySQL能听懂的语言。

我刚开始学的时候，也试过MySQLdb，但它在Python3上安装各种报错。后来发现pymysql完美支持Python3，安装简单，文档清晰，就成了我的首选。

1.2 一步步搭建环境

第一步：安装MySQL
如果你还没安装MySQL，我推荐用Docker，一键搞定：

# 拉取MySQL镜像dockerpullmysql:8.0# 运行MySQL容器dockerrun-d\--namemysql_dev\-p3306:3306\-eMYSQL_ROOT_PASSWORD=yourpassword\-eMYSQL_DATABASE=test_db\mysql:8.0

第二步：安装pymysql

pipinstallpymysql

第三步：验证安装
创建一个简单的测试脚本test_connection.py：

importpymysqltry:# 尝试连接数据库connection=pymysql.connect(host='localhost',port=3306,user='root',password='yourpassword',database='test_db',charset='utf8mb4')print("✅ 连接成功！")connection.close()exceptExceptionase:print(f"❌ 连接失败：{e}")

运行这个脚本，如果看到"✅ 连接成功！"，恭喜你，环境搭建完成！

二、核心概念：理解数据库连接的本质

2.1 连接对象 vs 游标对象

这是初学者最容易混淆的两个概念。让我用个比喻帮你理解：

• 连接对象：就像你去图书馆办了一张借书卡
• 游标对象：就像你拿着这张卡，在图书馆里找书、借书、还书

一个连接可以有多个游标，就像你可以同时借多本书一样。

importpymysql# 1. 建立连接（办借书卡）connection=pymysql.connect(host='localhost',user='root',password='yourpassword',database='test_db',charset='utf8mb4')# 2. 创建游标（开始找书）cursor=connection.cursor()# 3. 执行SQL（借书操作）cursor.execute("SELECT * FROM users")# 4. 获取结果（拿到书）results=cursor.fetchall()# 5. 关闭游标和连接（还书、退卡）cursor.close()connection.close()

2.2 为什么需要字符集设置？

我踩过的一个大坑：中文数据存到数据库变成乱码。原因就是字符集没设置对。

MySQL默认的latin1字符集不支持中文，我们需要显式指定utf8mb4：

• utf8mb4：真正的UTF-8，支持所有Unicode字符（包括emoji）
• utf8：MySQL的"伪UTF-8"，只支持基本多文种平面

# 正确的连接方式connection=pymysql.connect(charset='utf8mb4',# 必须设置！cursorclass=pymysql.cursors.DictCursor# 让结果以字典形式返回)

三、实战演练：完整的CRUD操作

让我们创建一个真实的用户管理系统，包含创建表、增删改查所有操作。

3.1 创建数据库表

首先，我们创建一个用户表。在实际项目中，我建议先用SQL工具（如MySQL Workbench）设计好表结构。

importpymysqldefcreate_users_table():"""创建用户表"""connection=pymysql.connect(host='localhost',user='root',password='yourpassword',database='test_db',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor)try:withconnection.cursor()ascursor:# 创建用户表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,password_hash VARCHAR(255) NOT NULL,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,is_active BOOLEAN DEFAULT TRUE)"""cursor.execute(sql)print("✅ 用户表创建成功")# 提交事务connection.commit()exceptExceptionase:print(f"❌ 创建表失败：{e}")connection.rollback()# 回滚事务finally:connection.close()# 执行创建表create_users_table()

3.2 插入数据（Create）

插入数据时，一定要使用参数化查询，这是防止SQL注入的关键！

defadd_user(username,email,password):"""添加新用户"""connection=pymysql.connect(host='localhost',user='root',password='yourpassword',database='test_db',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor)try:withconnection.cursor()ascursor:# 使用参数化查询防止SQL注入sql="""INSERT INTO users (username, email, password_hash)VALUES (%s,%s,%s)"""# 注意：这里使用元组传递参数cursor.execute(sql,(username,email,password))connection.commit()print(f"✅ 用户{username}添加成功，ID:{cursor.lastrowid}")returncursor.lastrowid# 返回自增IDexceptpymysql.err.IntegrityErrorase:print(f"❌ 添加失败：用户名或邮箱已存在")returnNoneexceptExceptionase:print(f"❌ 添加失败：{e}")connection.rollback()returnNonefinally:connection.close()# 测试添加用户add_user("张三","zhangsan@example.com","hashed_password_123")add_user("李四","lisi@example.com","hashed_password_456")

3.3 查询数据（Read）

查询是数据库最常用的操作。pymysql提供了几种获取结果的方式：

defquery_users():"""查询用户数据"""connection=pymysql.connect(host='localhost',user='root',password='yourpassword',database='test_db',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor)try:withconnection.cursor()ascursor:# 1. 查询所有用户cursor.execute("SELECT * FROM users")# 获取结果的几种方式：print("\n=== 所有用户 ===")all_users=cursor.fetchall()# 获取所有记录foruserinall_users:print(f"ID:{user['id']}, 用户名:{user['username']}, 邮箱:{user['email']}")# 2. 查询单个用户（带条件）cursor.execute("SELECT * FROM users WHERE username =%s",("张三",))single_user=cursor.fetchone()# 获取单条记录ifsingle_user:print(f"\n=== 单个用户查询 ===")print(f"找到用户:{single_user['username']}")# 3. 分页查询（实际项目常用）page=1page_size=10offset=(page-1)*page_sizecursor.execute("""SELECT * FROM usersWHERE is_active = TRUEORDER BY created_at DESCLIMIT%sOFFSET%s""",(page_size,offset))page_users=cursor.fetchall()print(f"\n=== 第{page}页用户（每页{page_size}条）===")foruserinpage_users:print(f"用户名:{user['username']}")exceptExceptionase:print(f"❌ 查询失败：{e}")finally:connection.close()# 执行查询query_users()

3.4 更新数据（Update）

更新操作需要特别注意事务处理，确保数据一致性。

defupdate_user_email(user_id,new_email):"""更新用户邮箱"""connection=pymysql.connect(host='localhost',user='root',password='yourpassword',database='test_db',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor)try:withconnection.cursor()ascursor:# 先检查用户是否存在cursor.execute("SELECT id FROM users WHERE id =%s",(user_id,))ifnotcursor.fetchone():print(f"❌ 用户ID{user_id}不存在")returnFalse# 更新邮箱sql="UPDATE users SET email =%sWHERE id =%s"affected_rows=cursor.execute(sql,(new_email,user_id))connection.commit()ifaffected_rows>0:print(f"✅ 用户{user_id}邮箱更新成功")returnTrueelse:print(f"⚠️ 用户{user_id}邮箱未变化")returnFalseexceptpymysql.err.IntegrityError:print(f"❌ 邮箱{new_email}已被其他用户使用")connection.rollback()returnFalseexceptExceptionase:print(f"❌ 更新失败：{e}")connection.rollback()returnFalsefinally:connection.close()# 测试更新update_user_email(1,"zhangsan_new@example.com")

3.5 删除数据（Delete）

删除操作要特别小心！在实际项目中，我们通常使用"软删除"（标记删除）而不是物理删除。

defdelete_user(user_id,soft_delete=True):"""删除用户（支持软删除）"""connection=pymysql.connect(host='localhost',user='root',password='yourpassword',database='test_db',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor)try:withconnection.cursor()ascursor:ifsoft_delete:# 软删除：只是标记为不活跃sql="UPDATE users SET is_active = FALSE WHERE id =%s"operation="禁用"else:# 物理删除：真正从数据库删除（慎用！）sql="DELETE FROM users WHERE id =%s"operation="删除"affected_rows=cursor.execute(sql,(user_id,))connection.commit()ifaffected_rows>0:print(f"✅ 用户{user_id}{operation}成功")returnTrueelse:print(f"❌ 用户{user_id}不存在")returnFalseexceptExceptionase:print(f"❌{operation}失败：{e}")connection.rollback()returnFalsefinally:connection.close()# 测试删除（使用软删除）delete_user(2,soft_delete=True)

四、高级技巧：事务处理和错误处理

4.1 为什么需要事务？

想象一下银行转账：从A账户扣款，向B账户加款。如果扣款成功但加款失败，钱就消失了！事务就是确保这两个操作要么都成功，要么都失败。

deftransfer_money(from_user_id,to_user_id,amount):"""转账操作（演示事务）"""connection=pymysql.connect(host='localhost',user='root',password='yourpassword',database='test_db',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor)try:withconnection.cursor()ascursor:# 开始事务connection.begin()# 1. 检查转出账户余额（假设有balance字段）cursor.execute("SELECT balance FROM accounts WHERE user_id =%s",(from_user_id,))from_balance=cursor.fetchone()['balance']iffrom_balance<amount:raiseValueError("余额不足")# 2. 扣款cursor.execute("UPDATE accounts SET balance = balance -%sWHERE user_id =%s",(amount,from_user_id))# 3. 加款cursor.execute("UPDATE accounts SET balance = balance +%sWHERE user_id =%s",(amount,to_user_id))# 4. 记录交易日志cursor.execute("""INSERT INTO transactions (from_user_id, to_user_id, amount, type)VALUES (%s,%s,%s, 'transfer')""",(from_user_id,to_user_id,amount))# 提交事务（所有操作都成功才执行）connection.commit()print(f"✅ 转账成功：{from_user_id}->{to_user_id}金额：{amount}")returnTrueexceptExceptionase:# 任何一步出错，回滚所有操作connection.rollback()print(f"❌ 转账失败，已回滚：{e}")returnFalsefinally:connection.close()

4.2 完整的错误处理模板

这是我多年总结的错误处理最佳实践：

defsafe_database_operation():"""安全的数据库操作模板"""connection=Nonetry:connection=pymysql.connect(host='localhost',user='root',password='yourpassword',database='test_db',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor,autocommit=False# 手动控制事务)withconnection.cursor()ascursor:# 你的数据库操作代码cursor.execute("SELECT * FROM users")results=cursor.fetchall()connection.commit()returnresultsexceptpymysql.err.OperationalErrorase:# 连接错误：网络问题、数据库宕机等print(f"数据库连接错误：{e}")ifconnection:connection.rollback()returnNoneexceptpymysql.err.IntegrityErrorase:# 完整性错误：唯一约束冲突、外键约束等print(f"数据完整性错误：{e}")ifconnection:connection.rollback()returnNoneexceptpymysql.err.DataErrorase:# 数据错误：数据类型不匹配、数据过长等print(f"数据错误：{e}")ifconnection:connection.rollback()returnNoneexceptExceptionase:# 其他未知错误print(f"未知错误：{e}")ifconnection:connection.rollback()returnNonefinally:# 确保连接被关闭ifconnection:connection.close()

五、实际项目应用：用户注册登录系统

让我们把这些知识整合到一个实际项目中：

classUserManager:"""用户管理类（封装数据库操作）"""def__init__(self):self.connection_params={'host':'localhost','user':'root','password':'yourpassword','database':'test_db','charset':'utf8mb4','cursorclass':pymysql.cursors.DictCursor,'autocommit':False}defregister_user(self,username,email,password):"""用户注册"""connection=pymysql.connect(**self.connection_params)try:withconnection.cursor()ascursor:# 检查用户名是否已存在cursor.execute("SELECT id FROM users WHERE username =%s",(username,))ifcursor.fetchone():return{"success":False,"message":"用户名已存在"}# 检查邮箱是否已存在cursor.execute("SELECT id FROM users WHERE email =%s",(email,))ifcursor.fetchone():return{"success":False,"message":"邮箱已注册"}# 插入新用户（实际项目中密码应该加密！）sql="""INSERT INTO users (username, email, password_hash)VALUES (%s,%s,%s)"""cursor.execute(sql,(username,email,password))user_id=cursor.lastrowid# 创建用户配置记录cursor.execute("INSERT INTO user_settings (user_id) VALUES (%s)",(user_id,))connection.commit()return{"success":True,"message":"注册成功","user_id":user_id}exceptExceptionase:connection.rollback()return{"success":False,"message":f"注册失败：{str(e)}"}finally:connection.close()deflogin(self,username,password):"""用户登录"""connection=pymysql.connect(**self.connection_params)try:withconnection.cursor()ascursor:# 查询用户信息sql="""SELECT id, username, email, password_hash, is_activeFROM usersWHERE username =%s"""cursor.execute(sql,(username,))user=cursor.fetchone()ifnotuser:return{"success":False,"message":"用户不存在"}ifnotuser['is_active']:return{"success":False,"message":"账户已被禁用"}# 验证密码（实际项目中应该使用加密验证）ifuser['password_hash']!=password:return{"success":False,"message":"密码错误"}# 更新最后登录时间cursor.execute("UPDATE users SET last_login = NOW() WHERE id =%s",(user['id'],))connection.commit()return{"success":True,"message":"登录成功","user":{"id":user['id'],"username":user['username'],"email":user['email']}}exceptExceptionase:connection.rollback()return{"success":False,"message":f"登录失败：{str(e)}"}finally:connection.close()# 使用示例if__name__=="__main__":manager=UserManager()# 注册新用户result=manager.register_user("王五","wangwu@example.com","password123")print(result)# 用户登录result=manager.login("王五","password123")print(result)

六、学习总结与避坑指南

6.1 关键要点回顾

1. 连接管理：每次操作后记得关闭连接，避免连接泄漏
2. 参数化查询：永远不要拼接SQL字符串，防止SQL注入
3. 事务处理：相关操作要放在同一个事务中
4. 错误处理：针对不同类型的数据库错误进行适当处理
5. 字符集设置：始终使用utf8mb4支持中文和特殊字符

6.2 常见问题与解决方案

6.3 下一步学习方向

掌握了pymysql的基本操作后，你可以继续学习：

1. 连接池技术：使用DBUtils管理数据库连接，提高并发性能
2. ORM框架：学习SQLAlchemy，用面向对象的方式操作数据库
3. 异步操作：使用aiomysql进行异步数据库操作
4. 数据库设计：学习规范化、索引优化、查询优化

学习交流与进阶

恭喜你完成了Python MySQL的基本操作学习！这只是开始，后面还有连接池、ORM等更高效的方式等着你。

欢迎在评论区分享：

• 你在连接MySQL时遇到了哪些报错？
• 文中的示例代码运行成功了吗？
• 对于事务管理，你还有什么疑问？

我会为初学者提供针对性的解答。记得多敲代码，这是学数据库最好的方式！

推荐学习资源：

1. pymysql官方文档 - 最权威的学习资料
2. GitHub上的Python MySQL示例 - 实战代码参考
3. MySQL官方教程 - 数据库系统学习

下篇预告：
下一篇将分享《Python MySQL事务实战》，从转账异常到数据一致性，手把手教你避坑。

学习建议：编程学习就像练功，不动手永远停留在理论阶段。立刻打开终端试试吧！遇到问题不要怕，每个错误都是进步的机会。我在学习pymysql时，光是一个字符集问题就调试了一下午，但解决后的成就感是无价的。