psycopg3 高效批量插入与冲突处理：executemany 的正确实践

新闻中心 NEWS CENTER

您当前位置：首页 > 新闻中心 > 网络学院

psycopg3 高效批量插入与冲突处理：executemany 的正确实践

2025-11-11

浏览次数：次

返回列表

psycopg3 高效批量插入与冲突处理：executemany 的正确实践

本教程详细探讨了 `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 的可读性和可维护性。

千鹿Pr助手

智能Pr插件，融入众多AI功能和海量素材

128 查看详情千鹿Pr助手

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 模块的优势：