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]]
# 尝试执行：cursor.executemany(sql, 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}")

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