使用SQLAlchemy向Azure SQL批量插入数据未持久化问题

我之前碰到过一模一样的情况！SSMS里跑BULK INSERT完全正常，Python里执行没报错但数据就是没存下来，核心原因其实是SQLAlchemy默认的事务机制——SSMS默认自动提交事务，但SQLAlchemy的连接会自动开启一个事务，你如果不手动提交，操作就会在会话结束时回滚，数据自然不会持久化。

具体解决办法 #

1. 手动提交事务 #

执行完BULK INSERT之后，直接调用连接的commit()方法就行：

query = ''' BULK INSERT mytable FROM '%s.csv' WITH (DATA_SOURCE = 'blobStorage', FIELDTERMINATOR = ',', ROWTERMINATOR = '\n', FIRSTROW=2) ''' % report_date
connection.execute(query)
# 关键一步：提交事务
connection.commit()

2. 用上下文管理器自动处理事务 #

推荐用with语句管理连接，这样上下文退出时会自动帮你提交（如果没有异常的话），更安全：

from sqlalchemy import create_engine

engine = create_engine('你的Azure SQL连接字符串')
with engine.connect() as connection:
    query = ''' BULK INSERT mytable FROM '%s.csv' WITH (DATA_SOURCE = 'blobStorage', FIELDTERMINATOR = ',', ROWTERMINATOR = '\n', FIRSTROW=2) ''' % report_date
    connection.execute(query)
    # 也可以手动在这里commit，或者让上下文自动处理
    connection.commit()

3. 开启自动提交模式 #

如果你的场景里不需要事务控制，也可以在创建连接时直接开启自动提交：

connection = engine.connect().execution_options(isolation_level="AUTOCOMMIT")
query = ''' BULK INSERT mytable FROM '%s.csv' WITH (DATA_SOURCE = 'blobStorage', FIELDTERMINATOR = ',', ROWTERMINATOR = '\n', FIRSTROW=2) ''' % report_date
connection.execute(query)
# 这里不用手动commit，因为自动提交已经开启

额外验证小技巧 #

提交之后可以立刻查一下数据数量，确认是否插入成功：

count_result = connection.execute("SELECT COUNT(*) FROM mytable")
print(f"当前表中数据行数：{count_result.scalar()}")

既然SSMS里能成功，说明你的BULK INSERT语句和Blob数据源配置都没问题，只要搞定事务提交这一步，数据肯定能正常持久化啦！

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