Python操作MySQL数据库：从入门到实战的完整指南

MySQL作为全球最流行的开源关系型数据库之一，广泛应用于Web开发、数据分析、后台系统等场景。而Python凭借简洁的语法和丰富的库生态，成为操作MySQL的首选语言之一。本文将以实战为导向，详细讲解如何通过Python的pymysql库实现MySQL数据库的连接、增删改查（CRUD）、事务处理等核心操作，同时覆盖生产环境中的优化技巧和避坑要点，适合Python初学者和后端开发人员学习。

一、准备工作：环境搭建与依赖安装

1. 确认MySQL环境

首先确保本地或远程服务器已安装MySQL数据库，可通过以下命令验证（Windows需配置MySQL环境变量，Linux/macOS直接执行）：

# 登录MySQL终端mysql -u root -p

输入密码后能正常进入MySQL交互界面，说明环境可用。若未安装MySQL，可参考官方文档（MySQL下载地址）完成安装。

2. 安装pymysql库

pymysql是纯Python实现的MySQL客户端库，兼容Python 3.x，无需编译，安装简单：

# 基础安装pipinstallpymysql# 安装指定稳定版本（推荐）pipinstallpymysql==1.1.0

安装完成后，在Python脚本中导入验证，无报错即安装成功：

importpymysql

二、核心操作：连接MySQL数据库

1. 基础连接语法

pymysql通过connect()方法创建数据库连接对象，核心参数如下（必选参数标注*）：

参数名	说明	示例
host*	MySQL服务器地址	localhost/192.168.1.10
port	端口号，默认3306	3306
user*	数据库用户名	root
password*	数据库密码	123456
database*	要连接的数据库名	test_db
charset	字符编码，推荐utf8mb4	utf8mb4
connect_timeout	连接超时时间（秒）	10

2. 最简连接示例

importpymysql# 1. 创建连接conn=pymysql.connect(host="localhost",port=3306,user="root",password="你的MySQL密码",database="test_db",charset="utf8mb4")# 2. 创建游标（执行SQL的核心对象）cursor=conn.cursor()# 3. 验证连接print("数据库连接成功！")# 4. 关闭资源（避免内存泄漏）cursor.close()conn.close()

3. 进阶：使用with语句自动释放资源

手动关闭游标和连接容易遗漏，推荐使用with语句，代码块结束后自动释放资源：

importpymysql# with语句管理连接和游标withpymysql.connect(host="localhost",port=3306,user="root",password="你的MySQL密码",database="test_db",charset="utf8mb4")asconn:withconn.cursor()ascursor:# 执行SQL操作（后续示例均基于此写法）print("连接已建立，游标创建完成")

三、实战操作：数据库增删改查（CRUD）

先准备测试表

为方便演示，先在MySQL中创建user表：

CREATETABLEIFNOTEXISTSuser(idINTPRIMARYKEYAUTO_INCREMENT,nameVARCHAR(50)NOTNULL,ageINT,genderVARCHAR(10),create_timeDATETIMEDEFAULTCURRENT_TIMESTAMP)ENGINE=InnoDBDEFAULTCHARSET=utf8mb4;

1. 数据查询（SELECT）

查询是最常用的操作，pymysql提供fetchone()（单条）、fetchall()（全部）、fetchmany(n)（指定条数）三种获取结果的方式。

示例1：查询单条数据（返回字典格式）

importpymysqldefquery_single_user(user_id):withpymysql.connect(host="localhost",port=3306,user="root",password="你的MySQL密码",database="test_db",charset="utf8mb4")asconn:# 指定游标返回字典（key为字段名），更易读withconn.cursor(cursor=pymysql.cursors.DictCursor)ascursor:# 参数化查询（%s为占位符），避免SQL注入sql="SELECT id, name, age, gender FROM user WHERE id = %s"cursor.execute(sql,(user_id,))# 获取单条结果result=cursor.fetchone()ifresult:print(f"查询结果：ID={result['id']}，姓名={result['name']}，年龄={result['age']}")else:print("无匹配数据")# 调用函数查询ID为1的用户query_single_user(1)

示例2：查询多条数据

importpymysqldefquery_multi_users(min_age):withpymysql.connect(host="localhost",port=3306,user="root",password="你的MySQL密码",database="test_db",charset="utf8mb4")asconn:withconn.cursor(cursor=pymysql.cursors.DictCursor)ascursor:# 查询年龄大于指定值的用户sql="SELECT id, name, age FROM user WHERE age > %s"cursor.execute(sql,(min_age,))# 获取所有结果results=cursor.fetchall()print(f"共查询到{cursor.rowcount}条数据：")forrowinresults:print(f"-{row['name']}（{row['age']}岁）")# 调用函数查询年龄大于18的用户query_multi_users(18)

2. 数据插入（INSERT）

插入数据需注意提交事务（conn.commit()），否则数据不会写入数据库；批量插入推荐使用executemany()提升效率。

importpymysqldefinsert_user():withpymysql.connect(host="localhost",port=3306,user="root",password="你的MySQL密码",database="test_db",charset="utf8mb4")asconn:withconn.cursor()ascursor:# 单条插入sql_single="INSERT INTO user (name, age, gender) VALUES (%s, %s, %s)"cursor.execute(sql_single,("张三",22,"男"))# 批量插入（效率更高）user_list=[("李四",20,"女"),("王五",25,"男"),("赵六",19,"女")]cursor.executemany(sql_single,user_list)# 提交事务（关键！否则数据不生效）conn.commit()print(f"插入成功，共新增{cursor.rowcount}条数据")# 调用函数插入数据insert_user()

3. 数据更新（UPDATE）

importpymysqldefupdate_user_age(user_name,new_age):withpymysql.connect(host="localhost",port=3306,user="root",password="你的MySQL密码",database="test_db",charset="utf8mb4")asconn:withconn.cursor()ascursor:# 更新指定用户的年龄sql="UPDATE user SET age = %s WHERE name = %s"cursor.execute(sql,(new_age,user_name))# 提交事务conn.commit()ifcursor.rowcount>0:print(f"更新成功，共修改{cursor.rowcount}条数据")else:print("无匹配用户，未修改任何数据")# 调用函数将“张三”的年龄改为23update_user_age("张三",23)

4. 数据删除（DELETE）

删除数据需谨慎，建议先通过查询验证条件，再执行删除操作。

importpymysqldefdelete_user(age_limit):withpymysql.connect(host="localhost",port=3306,user="root",password="你的MySQL密码",database="test_db",charset="utf8mb4")asconn:withconn.cursor()ascursor:# 删除年龄小于指定值的用户sql="DELETE FROM user WHERE age < %s"cursor.execute(sql,(age_limit,))# 提交事务conn.commit()print(f"删除成功，共删除{cursor.rowcount}条数据")# 调用函数删除年龄小于18的用户delete_user(18)

四、高级技巧：事务与异常处理

1. 事务处理

MySQL默认开启事务，当执行多个关联操作时（如转账：扣减A账户+增加B账户），需保证全部操作成功才提交，任一失败则回滚，避免数据不一致。

importpymysqldeftransfer_money(from_id,to_id,amount):conn=Nonecursor=Nonetry:conn=pymysql.connect(host="localhost",port=3306,user="root",password="你的MySQL密码",database="test_db",charset="utf8mb4")cursor=conn.cursor()# 步骤1：扣减转出账户金额sql1="UPDATE account SET balance = balance - %s WHERE id = %s"cursor.execute(sql1,(amount,from_id))# 步骤2：增加转入账户金额sql2="UPDATE account SET balance = balance + %s WHERE id = %s"cursor.execute(sql2,(amount,to_id))# 提交事务（全部操作成功）conn.commit()print("转账成功！")exceptpymysql.MySQLErrorase:# 回滚事务（任一操作失败）conn.rollback()print(f"转账失败，已回滚：{e}")finally:# 关闭资源ifcursor:cursor.close()ifconn:conn.close()# 调用函数：从ID=1的账户转账100元到ID=2的账户transfer_money(1,2,100)

2. 异常处理

常见异常包括连接失败、SQL语法错误、主键冲突等，需捕获pymysql.MySQLError并针对性处理：

importpymysqldefsafe_query():try:withpymysql.connect(host="localhost",port=3306,user="root",password="错误密码",# 故意设置错误密码database="test_db",charset="utf8mb4")asconn:withconn.cursor()ascursor:cursor.execute("SELECT * FROM user")exceptpymysql.OperationalErrorase:print(f"连接错误：{e}")# 如密码错误、服务器未启动exceptpymysql.ProgrammingErrorase:print(f"SQL语法错误：{e}")# 如表名写错、字段不存在exceptpymysql.IntegrityErrorase:print(f"数据完整性错误：{e}")# 如主键重复、外键约束exceptpymysql.MySQLErrorase:print(f"MySQL通用错误：{e}")

五、生产环境优化要点

1. 防止SQL注入

绝对禁止直接拼接SQL字符串，必须使用参数化查询（%s占位符）。以下是危险示例（禁止使用）和正确示例对比：

# 危险！易被SQL注入（如name值为 "张三' OR 1=1 --"）name="张三"sql=f"SELECT * FROM user WHERE name = '{name}'"# 禁止！# 正确：参数化查询sql="SELECT * FROM user WHERE name = %s"cursor.execute(sql,(name,))# 安全

2. 使用连接池

频繁创建/关闭连接会消耗大量资源，生产环境推荐使用pymysqlpool（连接池库）：

# 安装连接池库pipinstallpymysqlpool

frompymysqlpoolimportConnectionPool# 配置连接池pool=ConnectionPool(host="localhost",port=3306,user="root",password="你的MySQL密码",database="test_db",charset="utf8mb4",max_size=10,# 最大连接数min_size=2# 最小空闲连接数)# 获取连接并执行操作defquery_with_pool():conn=pool.get_connection()withconn.cursor(cursor=pymysql.cursors.DictCursor)ascursor:cursor.execute("SELECT * FROM user LIMIT 5")print(cursor.fetchall())# 归还连接到池（无需关闭）conn.close()query_with_pool()

3. 字符编码优化

连接时指定charset="utf8mb4"（而非utf8），utf8mb4兼容所有Unicode字符（包括emoji），避免插入特殊字符时出现编码错误。

4. 批量操作优化

插入/更新大量数据时，优先使用executemany()而非循环execute()，减少网络交互次数；同时可关闭自动提交（conn.autocommit(False)），批量操作后一次性提交。

六、常见问题排查

1. Access denied for user：检查用户名/密码是否正确，或MySQL用户是否允许当前IP访问（执行GRANT ALL ON *.* TO 'root'@'%' IDENTIFIED BY '密码'; FLUSH PRIVILEGES;开放远程访问）。
2. TimeoutError：确认MySQL服务器已启动，防火墙/安全组开放3306端口，或增加connect_timeout参数。
3. 数据插入后查询不到：未执行conn.commit()提交事务，或连接的数据库名称错误。
4. 中文乱码：连接时未指定charset="utf8mb4"，或表的字符集不是utf8mb4。