Python+Psycopg2：如何高效跨不同Schema批量更新数据？

PostgreSQL中不同Schema的表是独立对象，无法通过单条SQL跨多个Schema完成批量更新，因此按Schema分组后结合execute_values批量执行是最高效的方案——既保留了你熟悉的批量操作效率，又能适配多Schema的场景。

具体实现步骤 #

1. 按Schema分组数据：把同属一个Schema的处理结果归为一组，避免重复操作不同Schema的表。
2. 复用数据库连接：用同一个连接处理所有分组，减少连接建立/销毁的开销。
3. 动态生成带Schema的SQL：对每个分组，拼接对应Schema的表名，再用execute_values批量更新。

修改后的代码示例 #

class Message:
    messageId: str
    schema: str

def process_messages(messages):
    results = []
    for m in messages:
        status = process(m)  # 远程接口调用
        results.append((status, m))
    write_statuses(results)
    return results

from psycopg2.extras import execute_values
import collections

def write_statuses(results):
    # 1. 按Schema分组数据：key=schema，value=[(status, messageId), ...]
    schema_groups = collections.defaultdict(list)
    for status, msg in results:
        schema_groups[msg.schema].append((status, msg.messageId))

    conn = psycopg2.connect()  # 填写你的连接参数
    cur = conn.cursor()

    try:
        # 2. 遍历每个Schema分组，执行批量更新
        for schema, args_list in schema_groups.items():
            # 动态拼接带Schema的表名，用标识符转义避免SQL注入
            table_name = f"{cur.mogrify('%I.%I', (schema, 't_process_status')).decode('utf-8')}"
            
            sql = f"""
                UPDATE {table_name}
                SET process_status = data.v1
                FROM (VALUES %s) as data(id, v1)
                WHERE notification_id::text = data.id
            """
            # 用execute_values批量执行，保持原有的page_size控制
            execute_values(cur, sql, args_list, template=None, page_size=100)
        
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        cur.close()
        conn.close()

关键细节说明 #

• SQL注入防护：用cur.mogrify('%I', ...)转义Schema和表名，避免因为Schema名称包含特殊字符或恶意输入导致的SQL注入。
• 效率保障：每个Schema只执行一次批量更新，比逐条操作效率提升明显，同时复用连接减少了IO开销。
• 事务控制：添加try/finally块保证事务提交或回滚，避免数据不一致。

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