使用psycopg2 execute_values执行UPDATE遇日期类型错误求助

问题场景 #

参照方案使用psycopg2的execute_values执行批量UPDATE操作时，遇到日期字段类型不匹配错误，相关代码及报错信息如下：

SQL更新语句 #

UPDATE_QUERY = """UPDATE SHARE_RETURN_DOC
                                      SET TO_DB_TS = v1,
                                          SUPERSEDED_DT = v2,
                                          FIRST_SEEN_DT = v3,
                                          LAST_SEEN_DT = v4,
                                          BATCH_ID = v5,
                                          DATA_PROC_ID = v6,
                                          CO_REG_DEBT = v7,
                                          DIR_DTL_CHG_IN = v8,
                                          SHRHLDR_LIST_CD = v9,
                                          SHRHLDR_LIST_CD = v10,
                                          SHRHLDR_LEGAL_STAT = v11,
                                          SHRHLDR_REFRESH_CD = v12,
                                          SHRHLDR_SUPRESS_IN = v13,
                                          BULK_LIST_ID = v14,
                                          DOC_TYPE_CD = v15,
                                          JACKET_NR = v16
                                     FROM (VALUES %s) u(id1, id2, v1, v2, v3, v4, v5, v6, v7, v8, 
                                                        v9, v10, v11, v12, v13, v14, v15, v16)                                                     
                                    WHERE SHARE_RETURN_DOC.REG_NB = u.id1
                                      AND SHARE_RETURN_DOC.ANN_RTN_DT = u.id2"""

Python执行方法 #

def tableUpdate(connection, cursor, query, dataframe):
    data = []
    dataframe = dataframe.toPandas()
    for x in dataframe.to_numpy():
        data.append(tuple(x))
    try:
        print(data)
        extras.execute_values(cursor, query, data)
        connection.commit()
    except (Exception, Error) as error:
        print("Error: %s" % error)
        connection.rollback()
        return 1
    finally:
        cursor.close()

错误信息 #

传入包含None值的批量数据时，报错：

column "superseded_dt" is of type date but expression is of type text
LINE 3: ...                                         SUPERSEDED_DT = v2,
                                                                    ^

错误原因 #

1. execute_values批量插入时，若数据中存在None，PostgreSQL无法自动推导对应字段类型，会将其识别为text类型，与目标表的date字段类型不兼容。
2. 之前尝试的v2::date[]属于逻辑错误，目标字段是单个日期而非日期数组，类型转换方向错误。

解决方案 #

以下两种方案任选其一即可解决问题：

方案1：在VALUES子句中明确指定字段类型 #

修改SQL语句，在临时表u的字段定义中，给日期类型字段明确指定date类型，让PostgreSQL直接识别正确类型：

UPDATE_QUERY = """UPDATE SHARE_RETURN_DOC
                                      SET TO_DB_TS = v1,
                                          SUPERSEDED_DT = v2,
                                          FIRST_SEEN_DT = v3,
                                          LAST_SEEN_DT = v4,
                                          BATCH_ID = v5,
                                          DATA_PROC_ID = v6,
                                          CO_REG_DEBT = v7,
                                          DIR_DTL_CHG_IN = v8,
                                          SHRHLDR_LIST_CD = v9,
                                          SHRHLDR_LIST_CD = v10,
                                          SHRHLDR_LEGAL_STAT = v11,
                                          SHRHLDR_REFRESH_CD = v12,
                                          SHRHLDR_SUPRESS_IN = v13,
                                          BULK_LIST_ID = v14,
                                          DOC_TYPE_CD = v15,
                                          JACKET_NR = v16
                                     FROM (VALUES %s) u(id1, id2, v1, v2 date, v3 date, v4 date, v5, v6, v7, v8, 
                                                        v9, v10, v11, v12, v13, v14, v15, v16)                                                     
                                    WHERE SHARE_RETURN_DOC.REG_NB = u.id1
                                      AND SHARE_RETURN_DOC.ANN_RTN_DT = u.id2"""

方案2：在SET子句中正确转换单个日期类型 #

将SUPERSEDED_DT = v2改为SUPERSEDED_DT = v2::date（注意不是数组），同时对其他日期字段FIRST_SEEN_DT、LAST_SEEN_DT做相同处理：

UPDATE_QUERY = """UPDATE SHARE_RETURN_DOC
                                      SET TO_DB_TS = v1,
                                          SUPERSEDED_DT = v2::date,
                                          FIRST_SEEN_DT = v3::date,
                                          LAST_SEEN_DT = v4::date,
                                          BATCH_ID = v5,
                                          DATA_PROC_ID = v6,
                                          CO_REG_DEBT = v7,
                                          DIR_DTL_CHG_IN = v8,
                                          SHRHLDR_LIST_CD = v9,
                                          SHRHLDR_LIST_CD = v10,
                                          SHRHLDR_LEGAL_STAT = v11,
                                          SHRHLDR_REFRESH_CD = v12,
                                          SHRHLDR_SUPRESS_IN = v13,
                                          BULK_LIST_ID = v14,
                                          DOC_TYPE_CD = v15,
                                          JACKET_NR = v16
                                     FROM (VALUES %s) u(id1, id2, v1, v2, v3, v4, v5, v6, v7, v8, 
                                                        v9, v10, v11, v12, v13, v14, v15, v16)                                                     
                                    WHERE SHARE_RETURN_DOC.REG_NB = u.id1
                                      AND SHARE_RETURN_DOC.ANN_RTN_DT = u.id2"""

额外优化：确保数据帧日期列类型正确 #

在转换为Pandas数据帧后，将日期列转换为date类型，让psycopg2自动适配PostgreSQL的日期类型，从根源避免类型推导问题：

def tableUpdate(connection, cursor, query, dataframe):
    dataframe = dataframe.toPandas()
    # 转换日期列为date类型，None会转为pd.NaT，psycopg2自动映射为PostgreSQL NULL
    date_cols = ['SUPERSEDED_DT', 'FIRST_SEEN_DT', 'LAST_SEEN_DT']
    for col in date_cols:
        dataframe[col] = pd.to_datetime(dataframe[col], errors='coerce').dt.date
    # 直接转换为元组列表，简化代码
    data = list(dataframe.itertuples(index=False, name=None))
    try:
        extras.execute_values(cursor, query, data)
        connection.commit()
    except (Exception, Error) as error:
        print("Error: %s" % error)
        connection.rollback()
        return 1
    finally:
        cursor.close()

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