使用SQLAlchemy和Pandas批量插入SQL Server报错求助

问题根源 #

你遇到的Expected 31 parameters, supplied 0错误，核心是参数传递格式与SQL占位符不匹配：

• 你用text()构造了带位置占位符?的SQL语句，但传递的是字典列表（df.to_dict(orient="records")），而?需要对应元组/列表形式的位置参数；
• 同时con.execute()默认单条执行，批量插入需明确启用批量模式。

两种可行解决方案 #

方案1：适配位置占位符，沿用元组列表参数（与原pyodbc用法兼容） #

调整参数格式为元组列表，同时启用SQL Server批量插入优化：

def insert_into_table(self, con: sqlalchemy.Connection, table: str, df: pd.DataFrame):
    # 构造插入语句
    columns = ",".join(df.columns)
    placeholders = ",".join(["?"] * len(df.columns))
    query = f"INSERT INTO {table} ({columns}) VALUES ({placeholders})"

    # 替换NaN为None避免插入错误
    df = df.replace({np.nan: None})
    # 转换为元组列表（和原pyodbc executemany参数格式一致）
    params = df.values.tolist()

    # 批量插入，启用fast_executemany优化SQL Server写入性能
    con.execute(
        sqlalchemy.text(query),
        params,
        execution_options={"fast_executemany": True}
    )

    con.commit()
    return

方案2：使用SQLAlchemy Core表对象（更安全，推荐） #

通过反射获取表结构，用Core语法避免手动拼接SQL，同时支持字典列表参数：

from sqlalchemy import MetaData, Table

def insert_into_table(self, con: sqlalchemy.Connection, table: str, df: pd.DataFrame):
    # 反射加载目标表的元数据（无需提前定义表结构）
    metadata = MetaData()
    target_table = Table(table, metadata, autoload_with=con)

    # 替换NaN为None
    df = df.replace({np.nan: None})
    params = df.to_dict(orient="records")

    # 批量插入
    con.execute(
        target_table.insert(),
        params,
        execution_options={"fast_executemany": True}
    )

    con.commit()
    return

关键说明 #

• fast_executemany是SQLAlchemy针对pyodbc连接SQL Server的专属优化，能大幅提升批量插入速度，效果等价于你之前用的pyodbcexecutemany；
• 方案2无需手动拼接SQL，避免了SQL注入风险，更符合SQLAlchemy的使用规范。

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