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

封神进阶！Python + SQL 高级玩法，批量操作+异常处理直接拉满

news 2026/4/24 3:20:56

封神进阶！Python + SQL 高级玩法，批量操作+异常处理直接拉满

家人们谁懂啊！上一篇 Python + SQL 增删改查火了之后，好多老铁私信我：“批量插10万条数据卡成狗咋整？”“报错了直接崩，数据乱成一锅粥怎么办？”

别急！新手入门看增删改查，高手进阶看批量优化+异常处理！今天咱依旧用幽默接地气的方式，把这两个高级技巧玩明白，让你写的代码又快、又稳、不报错，面试官看了都得夸一句“专业”！

先说好：本文全程无晦涩术语，主打一个“拿来就用”，所有代码可直接复制运行，结合上一篇的学生表案例，新手也能无缝衔接，建议收藏点赞，避免下次找不到！

先唠两句：为啥要学高级玩法？

上一篇教的基础批量操作、简单异常处理，应付小数据量（几百、几千条）没问题，但遇到这些场景就拉胯了：

批量插入10万条、100万条数据，循环插入卡到崩溃，等半天没反应；
代码报错后，数据要么丢一半、要么乱套，查错查到手软；
连接数据库后，遇到网络波动、数据库崩溃，程序直接挂掉，毫无还手之力；
批量修改/删除时，部分数据失败，导致整体操作功亏一篑。

而今天要讲的高级玩法，就是解决这些“痛点”——让批量操作飞起来，让异常报错可控，让你的代码从“能用”变成“好用、耐用”！

依旧用pymysql库，没安装的老铁先补票（上一篇的老粉可以跳过）：

pipinstallpymysql# 安装失败的话，换国内源（清华源解析失败可换其他源）pipinstallpymysql-ihttps://mirrors.aliyun.com/pypi/simple/

⚠️ 提示：上一篇提到的清华源（https://pypi.tuna.tsinghua.edu.cn/simple）目前网页解析失败，建议优先使用阿里云源，避免安装失败～

第一部分：批量操作优化（从“龟速”到“起飞”）

上一篇教的executemany\(\)虽然比循环单条插入快，但面对大数据量（1万条以上）还是不够看。今天教3个优化技巧，直接把效率拉满，亲测10万条数据插入从10分钟缩到1分钟！

先回顾：基础批量插入（反面教材，大数据量慎⽤）

上一篇的批量插入代码，适合小数据量，大数据量会卡顿，先放出来对比一下：

importpymysql# 基础连接（和上一篇一致，重点看批量插入部分）db=pymysql.connect(host='localhost',user='root',password='123456',database='test_db',charset='utf8mb4')cursor=db.cursor()# 基础批量插入（小数据量可用，大数据量卡顿）add_many_sql="INSERT INTO student(name, age, gender) VALUES(%s, %s, %s)"# 模拟1000条数据（大数据量可改成100000条，感受卡顿）students_data=[('学生'+str(i),18+i%3,'男'ifi%2==0else'女')foriinrange(1000)]cursor.executemany(add_many_sql,students_data)db.commit()print("基础批量插入完成！")# 关闭连接cursor.close()db.close()

❌ 痛点：每执行一次executemany\(\)，就会和数据库建立一次连接，大数据量时频繁交互，速度巨慢，还容易占用数据库资源。

优化技巧1：批量插入 + 事务批量提交（核心优化）

原理：把多个批量插入合并成一个事务，减少数据库连接次数，相当于“一次拉一车快递，比一次拉一个快10倍”！

重点：合理拆分数据（比如每1000条分一批），避免单次插入数据量太大，导致数据库报错。

importpymysql db=pymysql.connect(host='localhost',user='root',password='123456',database='test_db',charset='utf8mb4')cursor=db.cursor()# 优化：批量插入 + 事务批量提交add_many_sql="INSERT INTO student(name, age, gender) VALUES(%s, %s, %s)"# 模拟10万条数据（大数据量测试）students_data=[('学生'+str(i),18+i%3,'男'ifi%2==0else'女')foriinrange(100000)]# 拆分批次：每1000条分一批，避免单次插入过多batch_size=1000try:foriinrange(0,len(students_data),batch_size):batch_data=students_data[i:i+batch_size]cursor.executemany(add_many_sql,batch_data)# 所有批次执行完，统一提交事务（核心！减少连接次数）db.commit()print("10万条数据批量插入完成！速度飞起～")exceptExceptionase:db.rollback()print(f"插入失败：{e}")cursor.close()db.close()

✅ 优势：原本10万条数据要10分钟，优化后1分钟内搞定，而且减少数据库压力，不容易报错。

📌 小技巧：批次大小可调整（500-2000条为宜），根据自己的数据库性能灵活修改。

优化技巧2：关闭自动提交（锦上添花）

原理：pymysql 默认是自动提交（autocommit=True），每执行一次SQL就自动提交，大数据量时会拖慢速度，关闭后手动批量提交，进一步提升效率。

importpymysql# 优化2：关闭自动提交（在连接时设置autocommit=False）db=pymysql.connect(host='localhost',user='root',password='123456',database='test_db',charset='utf8mb4',autocommit=False# 关闭自动提交，手动控制事务)cursor=db.cursor()add_many_sql="INSERT INTO student(name, age, gender) VALUES(%s, %s, %s)"students_data=[('学生'+str(i),18+i%3,'男'ifi%2==0else'女')foriinrange(100000)]batch_size=1000try:foriinrange(0,len(students_data),batch_size):batch_data=students_data[i:i+batch_size]cursor.executemany(add_many_sql,batch_data)db.commit()print("关闭自动提交+批量提交，效率再提升！")exceptExceptionase:db.rollback()print(f"插入失败：{e}")cursor.close()db.close()

✅ 补充：关闭自动提交后，所有增删改操作都必须手动db\.commit\(\)，否则数据存不进去，别忘啦！

优化技巧3：批量修改/删除（避免循环操作）

新手误区：批量修改/删除时，用循环遍历每条数据执行SQL，速度巨慢！正确做法：用SQL的 IN 关键字，一次性执行。

importpymysql db=pymysql.connect(host='localhost',user='root',password='123456',database='test_db',charset='utf8mb4',autocommit=False)cursor=db.cursor()# 反面教材：循环修改（慢！不推荐）# for id in [1,2,3,4,5]:# update_sql = f"UPDATE student SET age=20 WHERE id={id}"# cursor.execute(update_sql)# 优化：批量修改（用IN关键字，一次性执行）update_sql="UPDATE student SET age=20 WHERE id IN (%s, %s, %s, %s, %s)"cursor.execute(update_sql,(1,2,3,4,5))# 传入需要修改的id列表# 批量删除同理（用IN关键字）delete_sql="DELETE FROM student WHERE id IN (%s, %s, %s)"cursor.execute(delete_sql,(6,7,8))db.commit()print("批量修改+删除完成，效率拉满！")cursor.close()db.close()

✅ 重点：IN 关键字后面的参数，用元组传入，避免拼接SQL字符串（防止SQL注入，后面会讲）。

第二部分：异常处理进阶（从“报错崩掉”到“可控可查”）

上一篇教的异常处理，只是简单的 try-except 捕获错误，不够全面。实际开发中，报错类型有很多（连接失败、SQL语法错误、数据重复、权限不足等），我们需要精准捕获、优雅处理，还能记录错误日志，方便排查。

核心目标：报错不崩程序、数据不混乱、错误能排查！

进阶技巧1：精准捕获不同类型的异常

pymysql 有很多专属异常（比如连接失败、主键重复），精准捕获这些异常，能快速定位问题，而不是只知道“报错了”。

importpymysql# 导入pymysql的异常类（重点！）frompymysqlimportOperationalError,ProgrammingError,IntegrityError db=Nonecursor=Nonetry:# 尝试连接数据库db=pymysql.connect(host='localhost',user='root',password='123456',# 故意填错密码，测试异常database='test_db',charset='utf8mb4')cursor=db.cursor()# 尝试执行SQL（故意写错表名，测试语法错误）sql="INSERT INTO student_error(name, age, gender) VALUES('测试', 18, '男')"cursor.execute(sql)db.commit()print("操作成功！")# 精准捕获异常：数据库连接失败（比如密码错误、主机不可达）exceptOperationalErrorase:print(f"数据库连接失败！错误原因：{e}")# 连接失败，无需回滚（还没连接上，没有事务）# 精准捕获异常：SQL语法错误（比如表名写错、字段写错）exceptProgrammingErrorase:print(f"SQL语法错误！错误原因：{e}")ifdb:# 若连接成功，才需要回滚db.rollback()# 精准捕获异常：数据完整性错误（比如主键重复、非空字段为空）exceptIntegrityErrorase:print(f"数据错误！错误原因：{e}")ifdb:db.rollback()# 捕获其他所有异常（兜底，避免程序崩掉）exceptExceptionase:print(f"未知错误！错误原因：{e}")ifdb:db.rollback()# 最终无论成功还是失败，都关闭连接（养成好习惯）finally:ifcursor:cursor.close()ifdb:db.close()print("连接已关闭！")

✅ 重点：

导入 pymysql 的异常类，才能精准捕获；
finally 块：无论是否报错，都会执行，确保连接关闭，避免资源泄露；
不同异常不同处理，比如连接失败不需要回滚，语法错误需要回滚。

进阶技巧2：记录错误日志（排查问题的“神器”）

实际开发中，只打印错误信息不够，需要把错误日志保存到文件，方便后续排查（比如线上代码报错，能快速找到问题所在）。

用 Python 内置的logging模块，简单易上手：

importpymysqlfrompymysqlimportOperationalError,ProgrammingError,IntegrityErrorimportlogging# 配置日志：保存到文件，格式包含时间、错误级别、错误信息logging.basicConfig(filename='sql_error.log',# 日志文件名称level=logging.ERROR,# 只记录错误级别及以上的日志format='%(asctime)s - %(levelname)s - %(message)s'# 日志格式)db=Nonecursor=Nonetry:db=pymysql.connect(host='localhost',user='root',password='123456',database='test_db',charset='utf8mb4')cursor=db.cursor()# 故意制造主键重复错误（假设id=1已存在）sql="INSERT INTO student(id, name, age, gender) VALUES(1, '测试', 18, '男')"cursor.execute(sql)db.commit()print("操作成功！")exceptOperationalErrorase:error_msg=f"数据库连接失败：{e}"print(error_msg)logging.error(error_msg)# 记录错误日志exceptProgrammingErrorase:error_msg=f"SQL语法错误：{e}"print(error_msg)logging.error(error_msg)ifdb:db.rollback()exceptIntegrityErrorase:error_msg=f"数据完整性错误：{e}"print(error_msg)logging.error(error_msg)ifdb:db.rollback()exceptExceptionase:error_msg=f"未知错误：{e}"print(error_msg)logging.error(error_msg)ifdb:db.rollback()finally:ifcursor:cursor.close()ifdb:db.close()print("连接已关闭！")

✅ 效果：运行后，错误信息会同时打印在控制台，还会保存到sql\_error\.log文件中，下次排查问题直接打开日志，一目了然。

进阶技巧3：防止SQL注入（安全第一！）

新手致命误区：拼接SQL字符串，比如sql = f\&\#34;SELECT \* FROM student WHERE name=\&\#39;\{name\}\&\#39;\&\#34;，一旦 name 是恶意数据（比如\&\#39; OR 1=1 \-\-），就会导致SQL注入，泄露甚至篡改数据！

正确做法：用参数化查询（%s 占位符），避免拼接字符串，pymysql 会自动处理恶意数据。

importpymysql db=pymysql.connect(host='localhost',user='root',password='123456',database='test_db',charset='utf8mb4')cursor=db.cursor()# 反面教材：拼接SQL字符串（危险！易被SQL注入）# name = "' OR 1=1 --" # 恶意数据# sql = f"SELECT * FROM student WHERE name='{name}'"# cursor.execute(sql)# 执行后会查询出所有学生数据，造成数据泄露# 正确做法：参数化查询（用%s占位符，传入元组）name="' OR 1=1 --"# 恶意数据，pymysql会自动处理sql="SELECT * FROM student WHERE name=%s"cursor.execute(sql,(name,))# 注意：参数必须是元组，即使只有一个参数result=cursor.fetchall()print("查询结果：",result)# 不会查询出所有数据，避免注入cursor.close()db.close()

🔴 警告：所有涉及用户输入、动态参数的SQL语句，都必须用参数化查询，别嫌麻烦，安全第一！

第三部分：高级玩法综合案例（实战必备）

把上面的优化技巧和异常处理结合起来，写一个完整的实战案例：批量插入10万条数据，带日志记录、异常捕获、防止注入，直接套用在项目中！

importpymysqlfrompymysqlimportOperationalError,ProgrammingError,IntegrityErrorimportlogging# 配置日志logging.basicConfig(filename='sql_advanced.log',level=logging.ERROR,format='%(asctime)s - %(levelname)s - %(message)s')defbatch_insert_students(students_data,batch_size=1000):""" 批量插入学生数据（优化版） :param students_data: 学生数据列表（列表嵌套元组） :param batch_size: 每批次插入数量 """db=Nonecursor=Nonetry:# 连接数据库，关闭自动提交db=pymysql.connect(host='localhost',user='root',password='123456',database='test_db',charset='utf8mb4',autocommit=False)cursor=db.cursor()# 批量插入SQL（参数化查询，防止注入）insert_sql="INSERT INTO student(name, age, gender) VALUES(%s, %s, %s)"# 拆分批次插入foriinrange(0,len(students_data),batch_size):batch_data=students_data[i:i+batch_size]cursor.executemany(insert_sql,batch_data)# 统一提交事务db.commit()print(f"批量插入成功！共插入{len(students_data)}条数据")returnTrueexceptOperationalErrorase:error_msg=f"数据库连接失败：{e}"print(error_msg)logging.error(error_msg)exceptProgrammingErrorase:error_msg=f"SQL语法错误：{e}"print(error_msg)logging.error(error_msg)ifdb:db.rollback()exceptIntegrityErrorase:error_msg=f"数据错误：{e}"print(error_msg)logging.error(error_msg)ifdb:db.rollback()exceptExceptionase:error_msg=f"未知错误：{e}"print(error_msg)logging.error(error_msg)ifdb:db.rollback()finally:# 关闭连接ifcursor:cursor.close()ifdb:db.close()print("数据库连接已关闭")returnFalse# 测试：批量插入10万条数据if__name__=="__main__":# 模拟10万条学生数据students=[('学生'+str(i),18+i%3,'男'ifi%2==0else'女')foriinrange(100000)]# 调用批量插入函数batch_insert_students(students,batch_size=1500)