如何使用psycopg2在PostgreSQL中正确实现批量upsert多行数据？

两个方案的问题原因 #

• 第一个方案报错核心原因：你使用excluded.*作为更新的右值，excluded.*会返回对应行的全表所有列，如果你动态插入的列只是表的部分列，右值列数和左侧你指定的更新列数就会不匹配，触发列数不一致的语法错误；另外你直接字符串拼接表名、列名，遇到保留字、特殊字符时也会偶发语法错误。
• 第二个方案报错核心原因：PostgreSQL的ON CONFLICT DO UPDATE是针对每条触发冲突的单独行做更新，只需指定单行的更新规则，不支持传入多组值，你的写法完全不符合语法规范，且直接拼接SQL存在严重SQL注入风险，不可用于生产环境。

正确批量Upsert实现 #

我们基于psycopg2.extras.execute_values实现，既保证性能，又适配动态列的需求，同时避免SQL注入风险：

1. 依赖导入 #

import psycopg2
from psycopg2 import extras, sql

2. 动态Upsert封装函数 #

def bulk_upsert(cursor, table_name: str, insert_cols: list[str], conflict_cols: list[str], values: list[tuple]):
    # 格式化标识符，避免保留字、特殊字符问题，防止SQL注入
    table = sql.Identifier(table_name)
    insert_columns = sql.SQL(', ').join(map(sql.Identifier, insert_cols))
    conflict_columns = sql.SQL(', ').join(map(sql.Identifier, conflict_cols))
    # 生成更新逻辑，左右列严格对齐，避免列数不匹配问题
    update_set = sql.SQL('({insert_cols}) = ({excluded_cols})').format(
        insert_cols=insert_columns,
        excluded_cols=sql.SQL(', ').join(sql.Identifier('excluded', col) for col in insert_cols)
    )
    # 拼接完整Upsert语句
    upsert_sql = sql.SQL("""
        INSERT INTO {table} ({insert_columns})
        VALUES %s
        ON CONFLICT ({conflict_columns}) DO UPDATE
        SET {update_set}
        WHERE {table} IS DISTINCT FROM excluded
    """).format(
        table=table,
        insert_columns=insert_columns,
        conflict_columns=conflict_columns,
        update_set=update_set
    )
    # 执行批量操作
    extras.execute_values(cursor, upsert_sql, values)

3. 调用示例 #

# 示例参数，可根据实际动态调整
table_name = 'your_table'
insert_cols = ['col1', 'col2', 'col3'] # 动态变动的插入列
conflict_cols = ['col1'] # 主键/唯一约束列，支持多列联合约束
values = [(1, 'test1', 100), (2, 'test2', 200), (3, 'test3', 300)] # 元组列表

# 执行Upsert
bulk_upsert(cursor, table_name, insert_cols, conflict_cols, values)
# 提交事务
conn.commit()

方案优势 #

• 动态生成更新逻辑时，左右两侧的列数量、顺序完全和传入的插入列对齐，彻底避免列数不匹配的错误
• 所有表名、列名都通过sql.Identifier处理，不会因保留字、特殊字符触发语法错误
• 保留了WHERE 表名 IS DISTINCT FROM excluded逻辑，跳过无数据变化的无效更新，大幅提升性能
• 全程使用参数化查询，无SQL注入风险，批量执行性能远高于逐行插入

内容的提问来源于stack exchange，提问作者jack.py