如何使用psycopg2的命名参数实现多行数据批量插入？

问题描述 #

psycopg2官方文档仅提供了单行命名参数插入的示例：

cur.execute("""
    INSERT INTO some_table (id, created_at, updated_at, last_name)
    VALUES (%(id)s, %(created)s, %(created)s, %(name)s);
    """,
    {'id': 10, 'name': "O'Reilly", 'created': datetime.date(2020, 11, 18)}
)

尝试编写多行批量插入代码时，抛出如下异常：

cursor.execute(sql, values)
TypeError: list indices must be integers or slices, not str

原代码实现：

from dataclasses import dataclass
from typing import Dict, List

import psycopg2


@dataclass
class MyObject:
    job_id: str
    param_a: str
    param_b: str

    def to_dict(self) -> Dict[str, str]:
        return {"job_id": self.job_id, "param_a": self.param_a, "param_b": self.param_b}


class DatabaseConnection:
    """
    Wrapper class to create a postgresql database connection.
    Is a helper to open and close the connection.
    """

    def __init__(self, database: str = "dbname") -> None:
        self.__database = database
        self.__connection = None

    def __enter__(self):
        self.__connection = psycopg2.connect(
            database=self.__database,
            user="user",
            password="pasword",
            host="host",
            port="8000",
            sslmode="require",
        )
        return self.__connection

    def __exit__(self, exception_type, exception_value, exception_traceback):
        if self.__connection:
            self.__connection.close()


def insert_values_to_db(objects: List[MyObject]) -> bool:
    try:
        with DatabaseConnection() as connection:
            cursor = connection.cursor()
            sql = (
                "INSERT INTO dbtable"
                "(job_id, param_a, param_b) VALUES"
                "(%(job_id)s, %(param_a)s, %(param_b)s);"
            )
            values = [obj.to_dict() for obj in objects]
            cursor.execute(sql, values)
            connection.commit()
    except Exception as _:
        return False

    return True

解决方案 #

问题出在cursor.execute()仅支持单条数据的参数传入，当传入字典列表时，它会尝试用字符串索引列表，导致类型错误。要实现命名参数的多行批量插入，只需将cursor.execute()替换为cursor.executemany()——这个方法专门用于批量执行SQL，支持接收字典列表作为参数。

修改后的insert_values_to_db函数：

def insert_values_to_db(objects: List[MyObject]) -> bool:
    try:
        with DatabaseConnection() as connection:
            cursor = connection.cursor()
            sql = (
                "INSERT INTO dbtable"
                "(job_id, param_a, param_b) VALUES"
                "(%(job_id)s, %(param_a)s, %(param_b)s);"
            )
            values = [obj.to_dict() for obj in objects]
            # 替换为executemany处理批量参数
            cursor.executemany(sql, values)
            connection.commit()
    except Exception as _:
        return False

    return True

补充说明 #

• executemany()会遍历传入的字典列表，为每个字典执行一次INSERT语句，自动完成命名参数的映射。
• 这种方式既保留了命名参数的可读性，又实现了批量插入需求，同时psycopg2会自动处理参数转义，避免SQL注入风险。

内容的提问来源于stack exchange，提问作者Jacobo