Python pandas DataFrame写入GP报NaTType对象不可调用错误

• 列名不匹配：查询数据时SELECT的字段为lastrun_date，插入逻辑中却读取了DataFrame中不存在的parsed_ts列，读取到的无效对象是首次触发object is not callable报错的直接原因。
• 时间值调用逻辑错误：修改代码时直接在列值后追加括号传入格式字符串，相当于将单元格内的值作为函数调用。当单元格值为pandas时间类型的空值NaT（即Not a Time，时间类型的空标记）时，就会抛出'NaTType' object is not callable异常。即便单元格是正常的时间戳类型，也不支持timestamp(格式串)的调用方式，时间格式化必须通过.strftime()方法实现。
• 数据库连接逻辑错误：在逐行插入的for循环内部执行了conn.close()，第一条数据插入完成后数据库连接就被断开，后续插入操作必然触发连接失效错误，属于未暴露的隐藏逻辑bug。

• 对齐列名映射：如果目标表需要的字段名为parsed_ts，可以在查询SQL中直接加别名SELECT name,lastrun_date as parsed_ts FROM Table，或者读取完成后通过dataFrame = dataFrame.rename(columns={'lastrun_date': 'parsed_ts'})重命名列，保证代码中引用的列名和DataFrame实际列完全一致。
• 规范处理时间类型与空值：不需要逐行将时间值转为字符串，批量预处理时间列即可：将列中的NaT空值替换为None，psycopg2会自动将None映射为数据库的NULL值，正常的pandas Timestamp类型也会被psycopg2自动适配为数据库的DateTime类型，无需手动做格式转换。
• 调整连接生命周期：将事务提交、连接关闭的逻辑移到for循环外部，所有插入语句执行完成后再统一提交事务，出现异常时主动回滚，禁止在循环内部关闭数据库连接。优先使用executemany做批量插入，替代逐行单条插入的写法，写入效率会有数量级提升。
• 简化实现方案：可以直接使用pandas内置的to_sql方法完成写入，不需要手动编写插入SQL、管理游标和连接，pandas会自动适配各字段类型，大幅减少手写代码的出错概率。

原生psycopg2批量插入版本 #

import psycopg2
import pandas as pd
from sqlalchemy import create_engine

# 从源库读取数据
alchemyEngine = create_engine('postgresql+psycopg2://user_name:Host:5432/warehouse')
with alchemyEngine.connect() as dbConnection:
    # 直接在SQL中对齐列名
    script = '''SELECT name,lastrun_date as parsed_ts FROM Table'''
    dataFrame = pd.read_sql(script, dbConnection)

# 预处理时间列：将NaT替换为None，适配psycopg2的空值规则
dataFrame['parsed_ts'] = dataFrame['parsed_ts'].replace({pd.NaT: None})

# 连接目标GP库
conn = psycopg2.connect(
    host=hostname,
    dbname=database,
    user=username,
    password=pwd,
    port=port_id
)
curr = conn.cursor()

try:
    # 构造批量插入参数
    insert_params = [
        (str(row['name']), row['parsed_ts'])
        for _, row in dataFrame.iterrows()
    ]
    curr.executemany(
        '''INSERT INTO Schema.destination_table (name,last_run) VALUES (%s,%s)''',
        insert_params
    )
    # 所有插入完成后统一提交
    conn.commit()
except Exception as e:
    # 异常时回滚事务
    conn.rollback()
    raise e
finally:
    # 最终统一关闭游标和连接
    curr.close()
    conn.close()

pandas to_sql极简版本 #

import pandas as pd
from sqlalchemy import create_engine

# 源库读数据
source_engine = create_engine('postgresql+psycopg2://user_name:Host:5432/warehouse')
with source_engine.connect() as conn:
    df = pd.read_sql('''SELECT name,lastrun_date as last_run FROM Table''', conn)

# 直接写入目标库
target_engine = create_engine('postgresql+psycopg2://username:pwd@hostname:5432/database')
df.to_sql(
    name='destination_table',
    schema='Schema',
    con=target_engine,
    if_exists='append',
    index=False,
    method='multi'
)

注意：使用to_sql时需要确保DataFrame的列名和目标表的字段名完全一致，字段顺序不需要强制对齐。

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