Python FastAPI端点批量插入SQL Server数据方案求助（无法使用SQLAlchemy）

方案1：使用pyodbc的executemany批量执行 #

这是最直接的优化方式，避免逐行循环，一次性传递所有行数据给数据库执行，大幅减少网络往返次数：

# 提取DataFrame中需要插入的列，转换为列表的列表（每行对应一个子列表）
data_rows = new_df[['account_no', 'amount', 'description', 'eff_date', 'reference']].values.tolist()

# 定义插入语句
insert_query = """
INSERT INTO ras_import_temp(account_no, amount, [description], eff_date, reference)
VALUES (?, ?, ?, ?, ?)
"""

# 批量执行插入
cursor.executemany(insert_query, data_rows)
cursor.commit()

对于数千行数据，这个方法的效率会比逐行插入提升数倍到数十倍，完全能解决端点超时问题。

方案2：使用SQL Server的BULK INSERT（适合超大量数据） #

如果数据量接近上万行，用数据库原生的批量导入命令效率更高。需要先把DataFrame导出为CSV，再让SQL Server直接读取文件：

本地文件版（适用于SQL Server可访问本地文件的场景） #

import tempfile
import os

# 创建临时CSV文件
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.csv', encoding='utf-8') as temp_file:
    new_df[['account_no', 'amount', 'description', 'eff_date', 'reference']].to_csv(
        temp_file, index=False, sep=',', quotechar='"', line_terminator='\n'
    )
temp_file_path = temp_file.name

try:
    # 执行BULK INSERT，注意转义路径中的反斜杠
    bulk_query = f"""
    BULK INSERT ras_import_temp
    FROM '{temp_file_path.replace("\\", "\\\\")}'
    WITH (
        FORMAT = 'CSV',
        FIRSTROW = 2,  # 跳过CSV表头
        FIELDTERMINATOR = ',',
        ROWTERMINATOR = '\\n',
        QUOTED_IDENTIFIER = ON
    )
    """
    cursor.execute(bulk_query)
    cursor.commit()
finally:
    # 清理临时文件
    os.unlink(temp_file_path)

Azure SQL Database适配版 #

如果是Azure SQL Database，本地文件无法被数据库访问，需要把CSV上传到Azure Blob存储，再通过OPENROWSET导入：

# 替换为你的Blob SAS URL（需赋予读取权限）
blob_sas_url = "https://yourstorage.blob.core.windows.net/container/temp.csv?your-sas-token"

bulk_query = f"""
INSERT INTO ras_import_temp(account_no, amount, [description], eff_date, reference)
SELECT * FROM OPENROWSET(
    BULK '{blob_sas_url}',
    FORMAT = 'CSV',
    FIRSTROW = 2,
    FIELDTERMINATOR = ',',
    ROWTERMINATOR = '\\n'
) AS bulk_data
"""
cursor.execute(bulk_query)
cursor.commit()

方案3：优化Pandas的to_sql（若可尝试SQLAlchemy） #

如果之前SQLAlchemy崩溃是因为配置或内存问题，可以尝试启用fast_executemany参数，这会让Pandas自动使用批量插入逻辑：

from sqlalchemy import create_engine

# 创建引擎时启用fast_executemany，提升插入效率
engine = create_engine(
    "mssql+pyodbc://user:password@server/database?driver=ODBC+Driver+17+for+SQL+Server",
    fast_executemany=True
)

# 分块插入，避免内存溢出
new_df.to_sql(
    'ras_import_temp',
    engine,
    if_exists='append',
    index=False,
    chunksize=1000  # 可根据服务器内存调整大小
)

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