本文旨在解决psycopg3中`executemany`方法批量插入多行数据时，针对`values %s`占位符与`on conflict`子句结合使用时遇到的常见`programmingerror`。我们将探讨如何正确构建包含多个列的`values`子句，提供两种解决方案：一种是基于字符串拼接的动态占位符生成，另一种是利用`psycopg.sql`模块进行更安全、更专业的
sql语句
组合，确保数据高效插入并妥善处理冲突。
Psycopg3中executemany批量插入的挑战
在Psycopg3中，executemany方法是实现批量数据插入的推荐方式，它能够高效地执行多条相似的SQL语句。然而，与Psycopg2的execute_values不同，直接将SQL语句中的VALUES子句简单地写为VALUES %s，并期望它能自动展开为多列占位符，会导致ProgrammingError: the query has 1 placeholder but X parameters were passed。这是因为Psycopg3要求VALUES子句中的占位符数量必须与要插入的列数精确匹配。
例如，对于一个包含7列的表，如果尝试使用如下SQL和数据：
执行时会抛出ProgrammingError，因为VALUES %s只提供了一个占位符，而values列表中的每个子列表却提供了7个参数。为了解决这个问题，我们需要确保VALUES子句包含与列数相匹配的占位符。
解决方案一：动态构建VALUES子句 (字符串拼接)
最直接的方法是根据要插入的列数，动态生成形如(%s, %s, ..., %s)的VALUES子句。这种方法简单易懂，适用于SQL结构相对固定的场景。
核心思路：
获取数据列表中每行元素的数量，这代表了要插入的列数。
生成与列数相同数量的%s占位符，并用逗号连接。
将这些占位符用括号括起来，形成完整的VALUES子句。
将这个动态生成的VALUES子句替换到原始SQL模板中。
示例代码：
注意事项：
unity3d游戏开发之相机切换 中文WORD版
本文档主要讲述的是unity3d游戏开发之相机切换；希望对大家会有帮助；感兴趣的朋友可以过来看看
下载
这种方法简单有效，但在构建复杂SQL或防止SQL注入方面存在潜在风险。
如果列数可能变化，确保num_columns的计算是准确的。
解决方案二：使用psycopg.sql模块安全构建SQL (推荐)
对于更专业、更安全的SQL语句构建，Psycopg3提供了psycopg.sql模块。这个模块允许你以编程方式组合SQL片段，从而避免手动字符串拼接可能带来的SQL注入风险，并提高代码的可读性和可维护性。
核心思路：
使用sql.SQL对象封装SQL语句的静态部分。
使用sql.Placeholder()生成单个占位符对象。
利用sql.SQL(', ').join()方法将多个sql.Placeholder()对象连接起来，形成动态的占位符列表。
使用sql.SQL.format()方法将动态生成的占位符列表注入到SQL语句中。
示例代码：
优势：
安全性：
psycopg.sql模块可以有效防止SQL注入攻击，因为它将SQL结构和参数值分离处理。
可读性与可维护性：
对于复杂的SQL语句，使用此模块可以使代码结构更清晰，更易于理解和维护。
灵活性：
能够以编程方式动态构建SQL的各个部分，适应各种复杂的查询需求。
总结与注意事项
在Psycopg3中使用executemany进行批量插入并处理冲突时，关键在于正确构建VALUES子句的占位符。
占位符数量匹配：
确保VALUES子句中的%s占位符数量与你尝试插入的列数严格一致。一个%s代表一个参数，而不是一行或一个多列结构。
ON CONFLICT子句：
ON CONFLICT (key_) DO UPDATE SET ...是PostgreSQL中实现UPSERT（更新或插入）逻辑的标准方式，它与executemany和动态占位符的构建完美结合。
推荐使用psycopg.sql模块：
尽管字符串拼接可以解决问题，但psycopg.sql模块提供了更安全、更健壮、更专业的SQL构建方式。特别是在生产环境或处理动态SQL时，强烈推荐使用它来组合SQL语句，以提高代码质量和安全性。
通过以上两种方法，你可以有效地在Psycopg3中利用executemany实现高效的批量数据插入和冲突处理。
sql = """
INSERT INTO activities (type_, key_, a, b, c, d, e)
VALUES %s
ON CONFLICT (key_) DO UPDATE
SET
a = EXCLUDED.a,
b = EXCLUDED.b,
c = EXCLUDED.c,
d = EXCLUDED.d,
e = EXCLUDED.e
"""
values = [['type', 'key', None, None, None, None, None]] # 实际数据，每行7个元素
# cursor.executemany(sql, values)import psycopg
# 假设这是你的原始SQL模板，其中包含一个占位符用于VALUES子句
# 注意：这里我们使用一个格式化字符串占位符 {} 来替换 VALUES 子句
base_sql_template = """
INSERT INTO activities (type_, key_, a, b, c, d, e)
VALUES {}
ON CONFLICT (key_) DO UPDATE
SET
a = EXCLUDED.a,
b = EXCLUDED.b,
c = EXCLUDED.c,
d = EXCLUDED.d,
e = EXCLUDED.e
"""
# 待插入的数据，每个子列表代表一行，包含7个元素
values_to_insert = [
['type1', 'key1', 1, 2, 3, 4, 5],
['type2', 'key2', 6, 7, 8, 9, 10],
['type3', 'key3', None, None, None, None, None]
]
if not values_to_insert:
print("没有数据可插入。")
else:
# 1. 获取列数（取第一行数据的长度）
num_columns = len(values_to_insert[0])
# 2. 生成占位符字符串，例如：'%s, %s, %s'
placeholders = ', '.join(['%s'] * num_columns)
# 3. 将占位符用括号括起来，形成 VALUES 子句，例如：'(%s, %s, %s)'
values_clause = f"({placeholders})"
# 4. 将 VALUES 子句注入到原始SQL模板中
final_sql = base_sql_template.format(values_clause)
print("生成的最终SQL语句示例:")
print(final_sql)
# 建立数据库连接并执行
try:
# 请替换为你的实际数据库连接信息
with psycopg.connect(dbname='test', user='your_user', password='your_password', host='localhost') as conn:
with conn.cursor() as cur:
cur.executemany(final_sql, values_to_insert)
conn.commit()
print(f"成功插入/更新 {len(values_to_insert)} 行数据。")
except psycopg.Error as e:
print(f"数据库操作失败: {e}")
import psycopg
from psycopg import sql
# 待插入的数据，每个子列表代表一行，包含7个元素
values_to_insert = [
['type1', 'key1', 1, 2, 3, 4, 5],
['type2', 'key2', 6, 7, 8, 9, 10],
['type3', 'key3', None, None, None, None, None]
]
if not values_to_insert:
print("没有数据可插入。")
else:
# 1. 获取列数
num_columns = len(values_to_insert[0])
# 2. 使用sql.Placeholder()生成与列数匹配的占位符列表
# sql.SQL(', ').join(...) 会将多个 sql.Placeholder() 用逗号连接
placeholders_sql = sql.SQL(', ').join(sql.Placeholder() * num_columns)
# 3. 构建完整的SQL语句，使用 {placeholders} 作为 VALUES 子句的占位符
# 注意：VALUES ({placeholders}) 中的括号是SQL语法的一部分
final_sql_obj = sql.SQL("""
INSERT INTO activities (type_, key_, a, b, c, d, e)
VALUES ({placeholders})
ON CONFLICT (key_) DO UPDATE
SET
a = EXCLUDED.a,
b = EXCLUDED.b,
c = EXCLUDED.c,
d = EXCLUDED.d,
e = EXCLUDED.e
""").format(placeholders=placeholders_sql) # 使用 .format() 注入动态生成的占位符
# 建立数据库连接并执行
try:
# 请替换为你的实际数据库连接信息
with psycopg.connect(dbname='test', user='your_user', password='your_password', host='localhost') as conn:
with conn.cursor() as cur:
# 打印生成的SQL语句（用于调试）
print("使用psycopg.sql生成的最终SQL语句示例:")
print(final_sql_obj.as_string(conn)) # as_string() 用于查看最终的SQL字符串
cur.executemany(final_sql_obj, values_to_insert)
conn.commit()
print(f"成功插入/更新 {len(values_to_insert)} 行数据。")
except psycopg.Error as e:
print(f"数据库操作失败: {e}")