一、应用场景
上篇文章我们学会了在pymysql事务中批量插入数据的复用代码，既然有了批量插入，那批量更新和批量删除的操作也少不了。
二、解决思路
为了解决批量删除和批量更新的问题，提出如下思路：
- 所有更新语句的生成都共用一个入口
- 更新的数据库字段和参数可动态决定
- 要操作的表名可动态决定
- 返回where条件前的sql语句
三、解决办法
- 这里只演示批量更新的解决办法，批量删除的类似。
- 跟插入语句不同的是，更新语句的where条件前set的数据才可以进行动态生成sql语句，而where条件后面的需要另外拼接。
- 例如：
update user_info set name ='头条号_code_space',size='30cm' where id=1 其中where条件前的sql就是我们要进行动态sql拼接的，通用方法此时只返回where条件前的sql语句。
def common_update_sql(item, table):
"""
从item的key-value提取要更新的字段，进行更新
:param item: 要执行更新的字段的字典，{"colunm_name_1":"colunm_value_1","colunm_name_2":"colunm_value_2"}
:param table: 数据库表名
:return: 返回拼接好的where条件前的sql更新语句
"""
update = ','.join([" {key} = %s".format(key=key) for key in item])
update_sql = 'update {table} set '.format(table=table) + update
return update_sql四、测试demo
# -*- coding: utf-8 -*-
"""
@Time ： 2022/1/29 12:50
@Auth ： 技术空间
@File ：handle_list_sql_demo2.py
@IDE ：PyCharm
@Motto：技术总是要日积月累的
"""
import pymysql
def common_update_sql(item, table):
"""
从item的key-value提取要更新的字段，进行更新
:param item: 要执行更新的字段的字典，{"colunm_name_1":"colunm_value_1","colunm_name_2":"colunm_value_2"}
:param table: 数据库表名
:return: 返回拼接好的where条件前的sql更新语句
"""
update = ','.join([" {key} = %s".format(key=key) for key in item])
update_sql = 'update {table} set '.format(table=table) + update
return update_sql
def select_list(db_cursor, sql):
"""
查询数据量表的列表
:param db_cursor: 游标
:param sql: 拼接好的sql查询语句
:return: 返回查询结果列表
"""
db_cursor.execute(sql)
data_list = db_cursor.fetchall()
print(data_list)
return data_list
if __name__ == '__main__':
db = pymysql.connect(host='localhost',
user='root',
password='root',
database='others')
cursor = db.cursor(pymysql.cursors.DictCursor)
table_1 = "user_info"
table_2 = "user_role"
select_user_sql = " select id,name from " + table_1
select_role_sql = " select user_id,role_id from " + table_2
try:
print("执行批量更新user前的数据-->")
select_list(cursor, select_user_sql)
update_user_list = []
for i in range(4, 7):
update_user_list.append({"name": "头条号_code_space_" + str(i)})
# 更新user表的id为4，5，6的数据
for i in range(3):
update = update_user_list[i]
update_sql = common_update_sql(update, table_1)
update_sql = update_sql + " where id=" + str(i + 4)
cursor.execute(update_sql, tuple(update.values()))
# 开始批量插入表2
print("执行批量更新user_role前的数据-->")
select_list(cursor, select_role_sql)
update_role_list = []
for i in range(4, 7):
update_role_list.append({"user_id": i, "role_id": 2})
# 更新user_role表的id为4，5，6的数据
for i in range(3):
update = update_role_list[i]
update_sql = common_update_sql(update, table_2)
update_sql = update_sql + " where id=" + str(i + 4)
cursor.execute(update_sql, tuple(update.values()))
except Exception as e:
# 事务回滚
db.rollback()
print('事务处理失败', e)
else:
# 事务提交
db.commit()
print('事务处理成功', cursor.rowcount)
print("执行批量更新user后的数据-->")
select_list(cursor, select_user_sql)
print("执行批量更新user_role后的数据-->")
select_list(cursor, select_role_sql)
cursor.close()
db.close()
关注我，坚持每日积累一个技巧，长期坚持，我们将会不断进步。