如何用Python的cx_Oracle实现Oracle表的动态INSERT操作？

需求 #

将Excel文件转成Pandas DataFrame后，匹配Oracle已存在表的同名列插入数据，要求后续数据集列扩展时无需修改代码；同时询问如何用CASE语句生成新列并插入。

现有代码 #

import pandas as pd
import numpy as np
import cx_Oracle

df = pd.read_excel("employee_info.xlsx")

con = None
try:
    con = cx_Oracle.connect (
          config.username,
          config.password,
          config.dsn,
          encoding = config.encoding)
except cx_Oracle.Error as error:
      print(error)
finally:
       cursor = con.cursor()
       rows = [tuple(x) for x in df.values]
       cursor.executemany( ''' INSERT INTO ODS.EMPLOYEES({x} VALUES {rows})   '''

已完成工作 #

已通过以下代码获取Excel与数据库表的共同列：

sql = "SELECT * FROM ODS.EMPLOYEES"
cursor.execute(sql)
data = cursor.fetchall()
col_names = []
for i in range (0, len(cursor.description)):
    col_names.append(cursor.description[i][0])

a = np.intersect1d(df.columns, col_names)
print("common columns:", a)

解决方案 #

1. 动态适配列的INSERT实现 #

核心思路是利用共同列列表动态生成SQL语句，确保DataFrame数据顺序与SQL列顺序一致，同时修复原代码中的错误：

import pandas as pd
import numpy as np
import cx_Oracle
import config  # 假设配置文件可正常导入

df = pd.read_excel("employee_info.xlsx")

con = None
cursor = None
try:
    # 建立数据库连接
    con = cx_Oracle.connect(
        config.username,
        config.password,
        config.dsn,
        encoding=config.encoding
    )
    cursor = con.cursor()

    # 高效获取数据库表列名（无需读取表数据）
    cursor.execute("SELECT * FROM ODS.EMPLOYEES WHERE 1=0")
    col_names = [desc[0] for desc in cursor.description]

    # 获取Excel与表的共同列
    common_cols = list(np.intersect1d(df.columns, col_names))
    if not common_cols:
        print("无匹配列，无法插入数据")
        exit()

    # 动态生成INSERT语句
    cols_str = ", ".join(common_cols)
    # 生成cx_Oracle兼容的位置占位符
    placeholders = ", ".join([f":{i+1}" for i in range(len(common_cols))])
    insert_sql = f"INSERT INTO ODS.EMPLOYEES ({cols_str}) VALUES ({placeholders})"

    # 按共同列顺序提取DataFrame数据，转为tuple列表
    rows = [tuple(row) for row in df[common_cols].values]

    # 批量插入并提交事务
    cursor.executemany(insert_sql, rows)
    con.commit()
    print(f"成功插入{len(rows)}条数据")

except cx_Oracle.Error as error:
    print(f"数据库错误: {error}")
    if con:
        con.rollback()
finally:
    # 关闭游标和连接
    if cursor:
        cursor.close()
    if con:
        con.close()

关键优化点：

• 用WHERE 1=0获取列名，避免读取表数据，提升效率
• 强制DataFrame按共同列顺序取数，避免列顺序不匹配导致的数据错误
• 完善事务回滚逻辑，保证数据一致性
• 动态生成SQL结构，后续列扩展时自动适配，无需修改代码

2. CASE语句生成新列的处理方式 #

两种方案可选，根据场景选择：

方案1：在INSERT语句中嵌入CASE逻辑

如果新列计算依赖数据库其他数据，或希望逻辑由数据库执行，可直接写入INSERT语句：

# 示例：新增IS_FULL_TIME列，根据EMPLOYEE_TYPE判断
common_cols = list(np.intersect1d(df.columns, col_names))
# 将新列加入目标列列表
target_cols = common_cols + ["IS_FULL_TIME"]
cols_str = ", ".join(target_cols)
# 占位符对应原数据列+CASE逻辑
placeholders = ", ".join([f":{i+1}" for i in range(len(common_cols))]) + ", CASE WHEN :{len(common_cols)+1} = '全职' THEN 'Y' ELSE 'N' END"
insert_sql = f"INSERT INTO ODS.EMPLOYEES ({cols_str}) VALUES ({placeholders})"

# 注意：需将用于判断的列数据加入rows中
rows = [tuple(row) + (row[df.columns.get_loc("EMPLOYEE_TYPE")],) for row in df[common_cols].values]

方案2：在Python中预处理DataFrame

如果新列仅依赖当前Excel数据，推荐在插入前用Pandas处理，更灵活且减少数据库压力：

# 示例：在DataFrame中生成新列
df["IS_FULL_TIME"] = df["EMPLOYEE_TYPE"].apply(lambda x: "Y" if x == "全职" else "N")

# 后续按动态逻辑处理即可，新列会自动参与匹配（若表中存在该列）

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