Psycopg3高效批量插入与冲突处理：executemany的正确姿势

本文旨在解决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和数据：

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)

执行时会抛出ProgrammingError，因为VALUES %s只提供了一个占位符，而values列表中的每个子列表却提供了7个参数。为了解决这个问题，我们需要确保VALUES子句包含与列数相匹配的占位符。

解决方案一：动态构建VALUES子句 (字符串拼接)

最直接的方法是根据要插入的列数，动态生成形如(%s, %s, ..., %s)的VALUES子句。这种方法简单易懂，适用于SQL结构相对固定的场景。

核心思路：

1. 获取数据列表中每行元素的数量，这代表了要插入的列数。
2. 生成与列数相同数量的%s占位符，并用逗号连接。
3. 将这些占位符用括号括起来，形成完整的VALUES子句。
4. 将这个动态生成的VALUES子句替换到原始SQL模板中。

示例代码：

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}")

注意事项：

Moshi Chat

法国AI实验室Kyutai推出的端到端实时多模态AI语音模型，具备听、说、看的能力，不仅可以实时收听，还能进行自然对话。

下载

• 这种方法简单有效，但在构建复杂SQL或防止SQL注入方面存在潜在风险。
• 如果列数可能变化，确保num_columns的计算是准确的。

解决方案二：使用psycopg.sql模块安全构建SQL (推荐)

对于更专业、更安全的SQL语句构建，Psycopg3提供了psycopg.sql模块。这个模块允许你以编程方式组合SQL片段，从而避免手动字符串拼接可能带来的SQL注入风险，并提高代码的可读性和可维护性。

核心思路：

1. 使用sql.SQL对象封装SQL语句的静态部分。
2. 使用sql.Placeholder()生成单个占位符对象。
3. 利用sql.SQL(', ').join()方法将多个sql.Placeholder()对象连接起来，形成动态的占位符列表。
4. 使用sql.SQL.format()方法将动态生成的占位符列表注入到SQL语句中。

示例代码：

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}")

优势：

• 安全性： psycopg.sql模块可以有效防止SQL注入攻击，因为它将SQL结构和参数值分离处理。
• 可读性与可维护性： 对于复杂的SQL语句，使用此模块可以使代码结构更清晰，更易于理解和维护。
• 灵活性： 能够以编程方式动态构建SQL的各个部分，适应各种复杂的查询需求。

总结与注意事项

在Psycopg3中使用executemany进行批量插入并处理冲突时，关键在于正确构建VALUES子句的占位符。

1. 占位符数量匹配： 确保VALUES子句中的%s占位符数量与你尝试插入的列数严格一致。一个%s代表一个参数，而不是一行或一个多列结构。
2. ON CONFLICT子句： ON CONFLICT (key_) DO UPDATE SET ...是PostgreSQL中实现UPSERT（更新或插入）逻辑的标准方式，它与executemany和动态占位符的构建完美结合。
3. 推荐使用psycopg.sql模块： 尽管字符串拼接可以解决问题，但psycopg.sql模块提供了更安全、更健壮、更专业的SQL构建方式。特别是在生产环境或处理动态SQL时，强烈推荐使用它来组合SQL语句，以提高代码质量和安全性。

通过以上两种方法，你可以有效地在Psycopg3中利用executemany实现高效的批量数据插入和冲突处理。