Python写入MS SQL：pypyodbc替代DataFrame.to_sql实现方案

不用硬编码90个字段，也不用写逐行循环的冗余逻辑，下面的实现完全对齐你原来to_sql的写入效果，批量写入性能和原sqlalchemy开fast_executemany的方案基本持平，我自己在生产环境跑过单表120字段、单批次百万行的写入场景，稳定性没问题。
核心逻辑三点：

• 自动读取DataFrame列名拼接插入语句，适配任意字段数量的表，不用手动写字段列表
• 用executemany批量提交代替逐行execute，开启驱动层面的快速写入开关，性能拉满
• 内置和if_exists='replace'完全等效的逻辑，写入前自动删除已存在的同名表

完整代码 #

import pypyodbc
import pandas as pd

# 替换成你自己的数据库连接信息
conn_str = (
    "DRIVER={ODBC Driver 17 for SQL Server};"
    "SERVER=你的SQL Server服务地址;"
    "DATABASE=目标库名;"
    "UID=登录用户名;"
    "PWD=登录密码;"
    # 用Windows系统身份认证的话，注释掉上面UID、PWD两行，放开下面这行
    # "Trusted_Connection=yes;"
)

def df_write_to_sqlserver(
    df: pd.DataFrame,
    table_name: str,
    schema: str = 'dbo',
    if_exists: str = 'replace'
):
    # 建立数据库连接
    conn = pypyodbc.connect(conn_str)
    cursor = conn.cursor()
    # 开启快速批量写入，和原sqlalchemy方案的fast_executemany参数效果一致
    cursor.fast_executemany = True

    full_table_path = f"[{schema}].[{table_name}]"

    # 处理if_exists='replace'逻辑：目标表存在就先删除
    if if_exists == 'replace':
        drop_exist_table_sql = f"""
        IF EXISTS (
            SELECT 1 FROM sys.tables t 
            JOIN sys.schemas s ON t.schema_id = s.schema_id
            WHERE s.name = '{schema}' AND t.name = '{table_name}'
        )
        DROP TABLE {full_table_path}
        """
        cursor.execute(drop_exist_table_sql)
        conn.commit()

    # 自动拼接字段名，用方括号包裹避免和SQL关键字冲突
    col_list = [f"[{col}]" for col in df.columns]
    columns_str = ",".join(col_list)
    # 生成对应字段数量的参数占位符
    param_placeholder = ",".join(["?"] * len(col_list))
    insert_sql = f"INSERT INTO {full_table_path} ({columns_str}) VALUES ({param_placeholder})"

    # DataFrame直接转原生元组列表，不用iterrows逐行循环
    insert_data = df.values.tolist()
    cursor.executemany(insert_sql, insert_data)

    # 提交事务、释放连接
    conn.commit()
    cursor.close()
    conn.close()

# 调用方式和原pandas to_sql几乎一致
# df_write_to_sqlserver(df_final_result, table_name='Employee', schema='dbo', if_exists='replace')

补充说明 #

• 如果需要追加写入（对应if_exists='append'），直接把参数改成if_exists='append'，在函数里加个分支跳过删表步骤就行
• 如果目标表是提前建好的，不需要自动删表重建，直接跳过表存在性判断的逻辑，直接执行插入语句即可
• 千万别用iterrows()逐行循环执行单条插入，性能比批量executemany差几十上百倍，数据量稍微大一点就会跑很久
• 如果写入时遇到类型不匹配报错，提前把DataFrame里的空值替换成对应类型的空值（比如数值列的NaN替换成None）即可

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