Psycopg3批量插入报错：替代execute_values遇编程问题求助

问题分析 #

你之前依赖Psycopg2的execute_values实现批量插入+冲突更新，但Psycopg3移除了该方法，直接用executemany报错的核心原因：

• Psycopg2的execute_values支持VALUES %s作为批量数据的占位符语法，Psycopg3完全不兼容这种写法。
• executemany的逻辑是重复执行单条SQL语句，要求SQL里的占位符数量与每条数据的字段数完全匹配。你原来的SQL仅留了1个占位符，但每条数据有7个字段，因此会触发参数数量不匹配的错误。

解决方案1：使用execute_batch（推荐） #

Psycopg3内置了execute_batch专门处理批量操作，用法简洁且性能优异，只需少量调整代码：

1. 调整SQL语句，将VALUES %s替换为对应字段数的占位符：

INSERT INTO activities (type_, key_, a, b, c, d, e)
VALUES (%s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (key_) DO UPDATE
SET
    a = EXCLUDED.a,
    b = EXCLUDED.b,
    c = EXCLUDED.c,
    d = EXCLUDED.d,
    e = EXCLUDED.e

2. 用execute_batch执行批量操作：

from psycopg import connect

# 建立连接与游标
conn = connect("你的数据库连接字符串")
cur = conn.cursor()

values = [['type', 'key', None, None, None, None, None], 
          # 可添加更多行数据
          ['type2', 'key2', 1, 2, 3, 4, 5]]

# 执行批量插入+冲突更新
cur.execute_batch(sql, values)

# 提交事务
conn.commit()
cur.close()
conn.close()

解决方案2：动态构建批量插入SQL #

如果需要生成单条包含所有数据的INSERT语句（如VALUES (row1),(row2)...），可借助Psycopg3的sql模块动态拼接占位符：

from psycopg import connect, sql

conn = connect("你的数据库连接字符串")
cur = conn.cursor()

values = [['type', 'key', None, None, None, None, None],
          ['type2', 'key2', 1, 2, 3, 4, 5]]

# 生成单条数据的占位符组：(%s,%s,%s,%s,%s,%s,%s)
single_row_placeholders = sql.Placeholder() * len(values[0])
# 生成所有行的占位符：(%s,...), (%s,...)
all_rows = sql.SQL(",").join([sql.SQL("({})").format(single_row_placeholders) for _ in values])

# 拼接完整SQL语句
final_sql = sql.SQL("""
    INSERT INTO activities (type_, key_, a, b, c, d, e)
    VALUES {all_rows}
    ON CONFLICT (key_) DO UPDATE
    SET
        a = EXCLUDED.a,
        b = EXCLUDED.b,
        c = EXCLUDED.c,
        d = EXCLUDED.d,
        e = EXCLUDED.e
""").format(all_rows=all_rows)

# 展平values为一维列表，传入execute执行
cur.execute(final_sql, [val for row in values for val in row])

conn.commit()
cur.close()
conn.close()

注意事项 #

• 无论采用哪种方案，必须提交事务，否则数据不会持久化到数据库。
• 若values包含大量数据，execute_batch会自动分批次执行，避免单条SQL过长引发性能问题，优先推荐使用该方案。

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