使用psycopg2 execute_values插入None至date字段报错的解决方法

问题描述 #

使用psycopg2的execute_values批量向PostgreSQL的product表插入数据时，尝试将None插入date类型的expiration_date字段，出现报错：column "expiration_date" is of type date but expression is of type text。此前在UPDATE语句中通过::date类型转换解决过类似问题，但该方法在INSERT中无效，求可行的解决方式。

原因分析 #

批量插入的VALUES子句中，未明确指定expiration_date的字段类型，PostgreSQL会将None解析为text类型，而目标字段是date类型，无法完成隐式类型转换，因此触发报错。UPDATE场景中可以通过::date解决，是因为SET操作的类型转换逻辑与INSERT的SELECT子句逻辑存在差异。

解决方法 #

方法1：在VALUES子表定义中指定字段类型 #

直接在products_insert的列定义里为expiration_date指定date类型，让PostgreSQL明确该列的类型，自动将None识别为date类型的NULL：

修改后的SQL语句：

sql = """
INSERT INTO product(
id, created_date_time, name, expiration_date
)
SELECT
id, NOW() AT TIME ZONE 'Asia/Seoul', name, expiration_date
FROM (VALUES %s)
AS products_insert (
    id,
    name,
    expiration_date date  -- 明确指定date类型
);
"""

方法2：在SELECT子句中进行类型转换 #

如果不想修改列定义，也可以在SELECT阶段对expiration_date做类型转换，NULL::date会被正确识别为date类型的空值：

修改后的SQL语句：

sql = """
INSERT INTO product(
id, created_date_time, name, expiration_date
)
SELECT
id, NOW() AT TIME ZONE 'Asia/Seoul', name, expiration_date::date  -- 添加类型转换
FROM (VALUES %s)
AS products_insert (
    id,
    name,
    expiration_date
);
"""

修改后的完整insert_bulk函数（方法1示例） #

def insert_bulk(products: list):
    try:
        sql = """
        INSERT INTO product(
        id, created_date_time, name, expiration_date
        )
        SELECT
        id, NOW() AT TIME ZONE 'Asia/Seoul', name, expiration_date
        FROM (VALUES %s)
        AS products_insert (
            id,
            name,
            expiration_date date
        );
        """

        products_insert = [[value for value in product.values()] for product in products]
        # products_insert示例: [['1','apple', None],['2', 'banana', None],['3', 'meat', None], ...]
        with Connector.connect() as connection:
            with connection.cursor() as cursor:
                psycopg2.extras.execute_values(
                    cursor,
                    sql,
                    products_insert,
                    page_size=500,
                )
                affected_row_count = cursor.rowcount
        print(affected_row_count, "products inserted successfully")
    except (Exception, Error) as error:
        print(
            "[product_repository] Error while insert_bulk",
            error,
        )

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