如何将Python cx_Oracle获取的结果集导入SQL Server数据表

核心原则：不需要把原生tuple结果集强行转list/dict，90%的结构不匹配问题都是字段顺序/类型没对齐导致的，和数据格式本身无关。

步骤1：先做字段对齐校验（最容易漏的一步） #

• 写Oracle查询语句时禁止用SELECT *，必须显式指定要同步的字段名，避免源表结构变动导致字段顺序错位。
• 提前确认SQL Server目标表的字段顺序、字段类型，和Oracle查询返回的字段一一对应：比如Oracle查的是(id, user_name, create_time, amount)，目标表插入时的字段顺序也必须完全一致，类型不兼容的提前在Oracle查询里用SQL函数转好（比如Oracle的Number类型对应SQL Server的int/decimal，Date类型对应datetime）。

步骤2：cx_Oracle取数（保留原生tuple格式即可） #

cx_Oracle默认返回的tuple结构是最适配数据库批量插入接口的格式，额外转list/dict属于多余操作，反而容易引入格式错误。
示例代码：

import cx_Oracle

# 初始化Oracle连接（替换成你自己的连接信息）
oracle_conn = cx_Oracle.connect("用户名/密码@OracleIP:端口/服务名")
oracle_cursor = oracle_conn.cursor()

# 显式指定查询字段，禁止SELECT *
query_sql = "SELECT id, user_name, create_time, amount FROM ORACLE_SOURCE_TABLE"
oracle_cursor.execute(query_sql)

# 直接拉取结果，返回的是(tuple1, tuple2, ...)格式，不需要额外转换
source_data = oracle_cursor.fetchall()
# 可选：获取查询字段名做校验
source_cols = [col[0] for col in oracle_cursor.description]

数据量超过10万行时不要用fetchall()一次性拉全量，改成每次fetchmany(1000)分批拉取分批插入，避免内存溢出。

步骤3：SQL Server端批量写入（推荐用pyodbc驱动） #

用参数化查询+executemany批量插入，不要逐行循环插入，不要自己拼接SQL字符串。
示例代码：

import pyodbc

# 初始化SQL Server连接（替换成你自己的连接信息）
mssql_conn = pyodbc.connect("DRIVER={ODBC Driver 17 for SQL Server};SERVER=SQLServerIP;DATABASE=库名;UID=用户名;PWD=密码")
mssql_cursor = mssql_conn.cursor()

# 目标表字段顺序必须和Oracle查询的字段顺序完全一致
target_table = "MSSQL_TARGET_TABLE"
target_cols = ["id", "user_name", "create_time", "amount"]
# 生成SQL Server参数占位符（用?）
placeholder_str = ",".join(["?"] * len(target_cols))
insert_sql = f"INSERT INTO {target_table} ({','.join(target_cols)}) VALUES ({placeholder_str})"

try:
    # executemany可以直接接收cx_Oracle返回的tuple格式结果集，不需要转换
    mssql_cursor.executemany(insert_sql, source_data)
    mssql_conn.commit()
    print(f"同步完成，共写入{len(source_data)}行数据")
except Exception as e:
    mssql_conn.rollback()
    print(f"写入失败已回滚，错误信息：{str(e)}")
finally:
    # 释放资源
    oracle_cursor.close()
    oracle_conn.close()
    mssql_cursor.close()
    mssql_conn.close()

转dict后结构不匹配的解决方法 #

如果你已经把数据转成了字典列表（每个元素是{"字段名":值}格式），不要直接把字典传入插入接口，驱动不会自动按字段名匹配值，需要先按目标字段顺序把值提取成tuple格式：

# 假设你已经转好的字典格式数据
dict_list = [{"id":1, "user_name":"张三", "create_time":"2024-01-01", "amount":99.9}]
# 按目标字段顺序拼接成tuple列表
formatted_data = [tuple(row[col] for col in target_cols) for row in dict_list]
# 再把formatted_data传入executemany即可

常见避坑点 #

• 不要逐行循环调用execute()插入，executemany批量插入的性能是逐行插入的几十上百倍。
• 日期、CLOB、BLOB这类特殊类型如果报转换错误，提前在Oracle查询阶段用TO_CHAR、TO_DATE等函数转成兼容格式，不要靠Python层硬转。
• 同步前可以先取1~2行数据做测试插入，确认字段映射没问题再跑全量。

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