如何用Python遍历ID列表并传入SQL Update Query批量更新数据

核心修改思路 #

将原手动输入单ID的逻辑，改为预定义目标ID列表，通过循环遍历列表自动完成「Database1查询VOICEMAIL→格式化字段→更新目标服务器table1」的完整流程，同时保留原代码的业务逻辑与技术实现细节。

完整代码实现 #

import pymysql
from pymysql.cursors import DictCursor

# 1. 预定义所有待处理的6位ID列表
target_ids = ["123456", "654321", "112233", "445566"]  # 替换为实际ID集合

# 2. Database1 连接配置（保留原连接参数）
db1_config = {
    'host': 'db1_server_ip',
    'user': 'db1_user',
    'password': 'db1_pwd',
    'database': 'Database1',
    'cursorclass': DictCursor
}

# 3. 目标数据库（另一服务器）连接配置（保留原连接参数）
target_db_config = {
    'host': 'target_server_ip',
    'user': 'target_user',
    'password': 'target_pwd',
    'database': 'target_db',
    'cursorclass': DictCursor
}

# 4. 保留原VOICEMAIL字段格式化逻辑（示例为去除空格+转大写，替换为实际业务规则）
def format_voicemail(voicemail_str):
    if not voicemail_str:
        return None
    return voicemail_str.strip().upper()

# 5. 遍历ID执行批量处理
for target_id in target_ids:
    try:
        # 查询Database1中的VOICEMAIL
        with pymysql.connect(**db1_config) as db1_conn:
            with db1_conn.cursor() as cursor:
                query_sql = "SELECT VOICEMAIL FROM source_table WHERE id = %s"
                cursor.execute(query_sql, (target_id,))
                result = cursor.fetchone()
                
                if not result:
                    print(f"ID {target_id} 在Database1中无匹配记录，跳过")
                    continue
                raw_voicemail = result['VOICEMAIL']
        
        # 格式化VOICEMAIL字段
        formatted_voicemail = format_voicemail(raw_voicemail)
        
        # 更新目标服务器的table1
        with pymysql.connect(**target_db_config) as target_conn:
            with target_conn.cursor() as cursor:
                update_sql = "UPDATE table1 SET voicemail = %s WHERE id = %s"
                cursor.execute(update_sql, (formatted_voicemail, target_id))
                target_conn.commit()
                print(f"ID {target_id} 处理完成")
    
    except Exception as e:
        print(f"处理ID {target_id} 时出错: {str(e)}")
        # 可选：将错误写入日志文件
        # with open('batch_process_error.log', 'a') as f:
        #     f.write(f"{datetime.datetime.now()}: ID {target_id} 错误信息: {str(e)}\n")

关键细节说明 #

• ID列表管理：将所有待处理ID集中存入target_ids，便于批量维护，可从文件/配置中心读取动态生成
• 连接资源管理：使用with上下文管理器自动处理数据库连接的打开与关闭，避免资源泄漏
• 容错机制：单个ID处理失败时捕获异常，不中断整个批量任务，同时输出错误信息便于排查
• 业务逻辑保留：原有的VOICEMAIL查询、格式化、更新逻辑完全保留，仅修改输入方式为批量遍历

内容的提问来源于stack exchange，提问作者seanboyd_2