优化Flask SQLAlchemy批量插入以提升效率

1. 合并删除操作，减少数据库交互 #

原代码循环删除每个月份的数据，会产生多次数据库往返请求，改成一次DELETE语句批量处理所有目标月份：

def delete_insert_data(df):
    months_to_erase = list(set(df['month']))
    year_to_erase = list(set(df['year']))[0]
    product_code = df.iloc[0,0]

    # 单条DELETE语句处理所有目标月份，替代循环
    db.session.execute(delete(finance_products).where(
        finance_products.product_code == product_code,
        finance_products.year == year_to_erase,
        finance_products.month.in_(months_to_erase)  # 使用in条件批量匹配
    ))

    records_to_insert = df.to_dict(orient='records')
    db.session.execute(insert(finance_products), records_to_insert)
    db.session.commit()

2. 用pandas to_sql配合fast_executemany=True优化插入 #

SQLAlchemy默认批量插入对SQL Server效率有限，改用pandas原生to_sql并开启fast_executemany（SQL Server专属优化参数），能大幅提升插入速度：

def delete_insert_data(df):
    months_to_erase = list(set(df['month']))
    year_to_erase = list(set(df['year']))[0]
    product_code = df.iloc[0,0]

    # 批量删除并提交
    db.session.execute(delete(finance_products).where(
        finance_products.product_code == product_code,
        finance_products.year == year_to_erase,
        finance_products.month.in_(months_to_erase)
    ))
    db.session.commit()

    # 用pandas to_sql执行批量插入
    engine = db.get_engine()
    df.to_sql(
        name=finance_products.__tablename__,
        con=engine,
        if_exists='append',
        index=False,
        fast_executemany=True  # 关键优化参数
    )

3. 添加复合索引加速删除查询 #

确保finance_products表在product_code、year、month上有复合索引，避免DELETE操作全表扫描：

CREATE NONCLUSTERED INDEX IX_finance_products_product_year_month 
ON finance_products (product_code, year, month);

4. 调整事务与数据库配置 #

• 复用数据库会话，避免每次函数调用重新创建会话
• 开启数据库的READ_COMMITTED_SNAPSHOT隔离级别，减少锁等待：

ALTER DATABASE YourDatabaseName SET READ_COMMITTED_SNAPSHOT ON;

5. 合并多轮操作减少事务提交 #

如果40次函数调用处理的是不同产品数据，可以收集所有待处理的product_code、year、month，一次性执行批量删除，再一次性插入所有数据，避免40次独立的事务提交。

内容的提问来源于stack exchange，提问作者Luiza Souza Simões