本教程详细探讨了 `psycopg3` 中使用 `executemany` 进行批量数据插入和冲突更新的正确方法。针对 `psycopg2` `execute_values` 的弃用，文章演示了如何构建动态 sql 语句以适应多行插入，重点讲解了占位符的正确配置，以及如何利用 `psycopg.sql` 模块提高 sql 语句构造的安全性和灵活性，避免常见的 `programmingerror`。
1. psycopg3 批量插入与 executemany 的挑战
在 psycopg2 中，execute_values 提供了一种便捷的方式来批量插入多行数据。然而，在 psycopg3 中，该方法已被移除，开发者需要转而使用 cursor.executemany()。使用 executemany 时，一个常见的误区是直接将 VALUES %s 用于表示多列的占位符，例如：
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]]
# 尝试执行：cursor.executemany(sql, values)这种做法会导致 ProgrammingError: the query has 1 placeholder but 7 parameters were passed 错误。这是因为 executemany 要求 SQL 语句中的占位符数量必须与每一行数据中的列数严格匹配，即每插入一列就需要一个 %s 占位符。VALUES %s 仅表示一个整体的占位符，而我们实际传入的 values 列表中的每个子列表包含多达7个元素。
正确的做法是为每一列提供一个 %s 占位符，并用括号将其包围，例如 VALUES (%s, %s, %s, ...)。
2. 方法一：通过字符串操作动态构建占位符
为了解决上述问题，我们可以根据待插入数据的列数，动态生成相应数量的占位符字符串。这种方法适用于列数不固定或需要在运行时确定的场景。
首先，确定数据中每行的列数。然后，生成与列数相同数量的 %s 占位符，并用逗号连接起来，最后用括号包裹形成 VALUES 子句。
import psycopg
# 示例数据，每行包含7列
values = [['type1', 'key1', 'val_a1', 'val_b1', 'val_c1', 'val_d1', 'val_e1'],
['type2', 'key2', 'val_a2', 'val_b2', 'val_c2', 'val_d2', 'val_e2'],
['type3', 'key3', None, None, None, None, None]]
# 假设所有行的列数相同，取第一行作为参考
num_columns = len(values[0])
# 生成占位符字符串，例如：(%s, %s, %s, %s, %s, %s, %s)
placeholders = ', '.join(['%s'] * num_columns)
values_clause = f"({placeholders})"
# 构建完整的 SQL 语句
# 注意：这里我们直接将占位符字符串注入到 SQL 模板中
sql_template = f"""
INSERT INTO activities (type_, key_, a, b, c, d, e)
VALUES {values_clause}
ON CONFLICT (key_) DO UPDATE
SET
a = EXCLUDED.a,
b = EXCLUDED.b,
c = EXCLUDED.c,
d = EXCLUDED.d,
e = EXCLUDED.e
"""
# 建立数据库连接并执行
try:
with psycopg.connect(dbname='your_database', user='your_user', password='your_password', host='localhost') as conn:
with conn.cursor() as cur:
cur.executemany(sql_template, values)
conn.commit()
print(f"成功插入/更新 {len(values)} 行数据。")
except psycopg.Error as e:
print(f"数据库操作失败: {e}")
注意事项：
- 这种方法虽然有效，但在拼接 SQL 语句时需要格外小心，以防范 SQL 注入风险，尤其当 values_clause 的内容并非完全由程序内部控制时。
- 对于复杂的动态 SQL 构建，字符串拼接可能导致代码可读性下降和维护困难。
3. 方法二：使用 psycopg.sql 模块构建安全动态 SQL
psycopg3 提供了 psycopg.sql 模块，这是一个更安全、更强大的工具，用于程序化地构建 SQL 语句。它能够帮助我们避免 SQL 注入风险，并提高动态 SQL 的可读性和可维护性。
psycopg.sql 模块的核心思想是将 SQL 语句的各个部分（如标识符、字面量、占位符）作为独立的 SQL 对象处理，然后通过 SQL 对象的 join、format 等方法进行组合。
import psycopg
from psycopg import sql
# 示例数据
values = [['type1', 'key1', 'val_a1', 'val_b1', 'val_c1', 'val_d1', 'val_e1'],
['type2', 'key2', 'val_a2', 'val_b2', 'val_c2', 'val_d2', 'val_e2'],
['type3', 'key3', None, None, None, None, None]]
num_columns = len(values[0])
# 使用 sql.Placeholder() 创建占位符对象
# sql.SQL(', ').join(...) 将占位符用逗号连接起来
placeholders = sql.SQL(', ').join(sql.Placeholder() * num_columns)
# 构建 SQL 语句模板，使用 {placeholders} 作为命名占位符
# 注意：这里的 VALUES ({placeholders}) 中的括号是 SQL 语法的一部分
isql_template = 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() 方法将占位符对象注入到 SQL 模板中
# psycopg.sql 会正确地处理这些占位符，生成安全的 SQL
final_isql = isql_template.format(placeholders=placeholders)
# 建立数据库连接并执行
try:
with psycopg.connect(dbname='your_database', user='your_user', password='your_password', host='localhost') as conn:
with conn.cursor() as cur:
# 可以打印生成的 SQL 语句以供调试
# print(f'Generated SQL: {final_isql.as_string(conn)}')
cur.executemany(final_isql, values)
conn.commit()
print(f"成功插入/更新 {len(values)} 行数据。")
except psycopg.Error as e:
print(f"数据库操作失败: {e}")
psycopg.sql 模块的优势：
- 安全性： 自动处理标识符和字面量的引用，有效防止 SQL 注入。
- 可读性： 将 SQL 结构化为 Python 对象，使动态 SQL 更易于理解和维护。
- 灵活性： 方便地组合复杂的 SQL 片段。
4. 总结
在 psycopg3 中进行批量数据插入和冲突更新时，executemany 是一个强大的工具。关键在于正确理解其占位符机制：对于 VALUES 子句，需要为每一列数据提供一个独立的 %s 占位符，并用括号包裹。
为了实现这一目标，我们可以选择：
- 字符串拼接： 简单直接，适用于占位符结构相对固定的场景，但需注意潜在的 SQL 注入风险。
- psycopg.sql 模块： 推荐用于构建更复杂、更安全的动态 SQL 语句，它通过对象化的方式管理 SQL 片段，提高了代码的健壮性和可维护性。
无论选择哪种方法，都应确保 SQL 语句的占位符数量与每行数据的列数精确匹配，这是避免 ProgrammingError 的核心。同时，合理利用事务管理，确保批量操作的原子性和数据一致性。
