使用cursor.executemany更新Oracle时如何绑定Pandas列到SQL变量

你触发的参数类型错误，原因是cursor.executemany()要求第二个参数传入批量参数的集合（列表/元组，每个元素对应单条SQL的入参），在循环中传入单行字典/序列不符合参数要求。
正确实现不需要逐行遍历DataFrame，直接按以下逻辑改造即可：

• 提前将DataFrame按SQL绑定变量的字段映射，一次性转换为字典列表作为批量入参
• 创建游标时开启batcherrors=True，执行过程中单条记录报错不会中断整体流程，所有错误会被自动收集
• 执行完成后通过cursor.getbatcherrors()获取全部失败记录，打印/写入日志即可，成功执行的记录正常提交生效

改造后完整代码 #

import datetime
import os
import cx_Oracle
import pandas as pd
# 补全原有config、log文件初始化等依赖导入

# SHRTCKN表更新SQL 原有语句无需修改
banner_shrtckn_update = """
UPDATE SATURN.SHRTCKN A
SET A.SHRTCKN_COURSE_COMMENT = :course_comment,
    A.SHRTCKN_REPEAT_COURSE_IND = :repeat_ind,
    A.SHRTCKN_ACTIVITY_DATE = SYSDATE,
    A.SHRTCKN_USER_ID = 'STU00940',
    A.SHRTCKN_DATA_ORIGIN = 'APPWORX'
WHERE A.SHRTCKN_PIDM = gb_common.f_get_pidm(:id) AND
      A.SHRTCKN_TERM_CODE = :term_code AND
      A.SHRTCKN_SEQ_NO = :seqno AND
      A.SHRTCKN_CRN = :crn AND
      A.SHRTCKN_SUBJ_CODE = :subj_code AND
      A.SHRTCKN_CRSE_NUMB = :crse_numb
"""

def writeLog(content):
    print(content)
    log.write(str(datetime.date.today())+" "+content+"\n")

def main():
    now = datetime.datetime.now()
    year = str(now.year)+"40"
    db_pass = os.environ['DB_PASSWORD']
    dsn = cx_Oracle.makedsn(host='FAKE', port='1521', service_name='TEST.FAKE.BLAH')
    banner_cnxn = None
    try:
        banner_cnxn = cx_Oracle.connect(user=config.db_test['user'], password = db_pass, dsn=dsn)
        writeLog("---- Oracle Connection Made ----")

        # 构造批量入参：映射DataFrame列和SQL绑定变量名
        col_mapping = {
            'id': 'Bear_Nbr',
            'term_code': 'Past_Term',
            'seqno': 'Seq_No',
            'crn': 'Past_CRN',
            'subj_code': 'Past_Prefix',
            'crse_numb': 'Past_Number',
            'course_comment': 'Past_Course_Comment',
            'repeat_ind': 'Past_Repeat_Ind'
        }
        # 重命名列后直接转字典列表，无需逐行循环
        param_df = df.rename(columns=col_mapping)[list(col_mapping.keys())]
        # 替换pandas空值NaN为Oracle识别的None
        param_list = param_df.where(pd.notnull(param_df), None).to_dict('records')
        total_rows = len(param_list)

        # 创建游标时开启batcherrors=True，允许跳过单条错误
        cursor = banner_cnxn.cursor()
        cursor.executemany(
            banner_shrtckn_update,
            param_list,
            batcherrors=True
        )

        # 收集所有错误记录写入日志
        errors = cursor.getbatcherrors()
        error_count = len(errors)
        for err in errors:
            # err.offset为错误行在参数列表中的索引（从0开始），err.message为具体错误信息
            err_row = param_list[err.offset]
            writeLog(f"更新失败，行参数：{err_row}，错误信息：{err.message}")

        # 提交成功的更新
        banner_cnxn.commit()
        success_count = total_rows - error_count
        writeLog(f"批量更新完成，共{total_rows}条，成功{success_count}条，失败{error_count}条")
        cursor.close()
    except Exception as e:
        # 此处仅捕获连接失败、SQL语法错误等全局级异常，单条数据错误不会触发
        writeLog(f"流程异常：{str(e)}")
    finally:
        if banner_cnxn:
            banner_cnxn.close()

if __name__ == "__main__":
    writeLog("-------- Process Start --------")
    main()
    writeLog("-------- Process End --------")

注意事项 #

• 该方案要求cx_Oracle版本≥7.0，版本过低先执行pip install --upgrade cx_Oracle升级
• 单条数据错误不会触发全局回滚，仅错误行不生效，其余正确行正常提交
• 如果数据量超过1万条，建议分批传入executemany（比如每1000条切分一次param_list），避免单次内存占用过高
• 如果需要关联原Excel行号排查错误，可以在构造param_list时提前保留原行号字段

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