Python写入SQL Server数据未持久化，本地查询却返回结果

我通过Python脚本调用API获取数据后，用SQLAlchemy连接SQL Server并写入数据，代码逻辑如下：

数据库连接代码 #

engine = create_engine(f'mssql+pyodbc://{sql_user}:{sql_password}@{sql_server}/{sql_database}?driver=ODBC+Driver+17+for+SQL+Server')
conn = engine.connect()
cursor = conn.connection.cursor()

数据存在性检查（首次查询返回None） #

find_query_string = f'SELECT * FROM [{table_name}] WHERE '
conditions = ' AND '.join([f'{col} = ?' for col in primary_key])
find_query_string += conditions
values = tuple([row_data[col] for col in primary_key])
print(find_query_string, values)
## SELECT * FROM [Timesheets] WHERE Id = ? AND UserId = ? (162019876, 1225205)
existing_row = cursor.execute(find_query_string, values).fetchone()
print(existing_row)
## None

数据插入代码 #

placeholder = ','.join(['?' for _ in range(len(columnn_mappings))])
insert_string = f'INSERT INTO [{table_name}] VALUES ({placeholder})'
insert_values = tuple(row_data.values())

print(insert_string, insert_values )
## INSERT INTO [Timesheets] VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) (162019876, 1225205, 15000169, '2024-01-23T07:55:00-08:00', '2024-01-23T11:55:42-08:00', Decimal('4.01'), '2024-01-23', -8, 'tsPT', 'regular', 'Kiosk (Surgery Ward)', 'False', 0, 'Forgot to clock out, Left at 5p on 1/22/23\nClocked in at 7:55a 1/23/24', Timestamp('2024-01-24 16:10:30+0000', tz='UTC'), 1225205)
cursor.execute(insert_string, insert_values )
conn.commit()

插入后查询（返回有效结果） #

existing_row = cursor.execute(find_query_string, values).fetchone()
print(existing_row)
## (162019876, 1225205, 15000169, '2024-01-23T07:55:00-08:00', '2024-01-23T11:55:42-08:00', Decimal('4.01'), '2024-01-23', -8, 'tsPT', 'regular', 'Kiosk (Surgery Ward)', 'False', 0, 'Forgot to clock out, Left at 5p on 1/22/23\nClocked in at 7:55a 1/23/24', datetime.datetime(2024, 1, 24, 16, 10, 30), 1225205)

但直接查看SQL Server数据库时，这条数据不存在，且SQL日志无错误信息。

核心原因 #

你混合使用了SQLAlchemy的Connection对象和底层pyodbc的Cursor，导致事务管理不一致：

• 底层Cursor执行的插入操作，属于底层连接的事务上下文
• 调用SQLAlchemy的conn.commit()时，SQLAlchemy未追踪到底层Cursor的操作，因此未真正提交该事务
• 脚本内的查询和插入处于同一个未提交事务中，所以能看到未提交的数据；但外部工具（如SSMS）使用默认的READ COMMITTED隔离级别，无法读取未提交数据，最终事务结束后数据自动回滚，数据库中无持久化记录。

解决方法 #

方法1：改用底层Cursor所属连接提交事务 #

将插入代码中的conn.commit()替换为cursor.connection.commit()，确保提交的是底层Cursor对应的事务：

cursor.execute(insert_string, insert_values )
cursor.connection.commit()  # 替换原conn.commit()

方法2：全程使用SQLAlchemy原生接口 #

避免直接操作底层Cursor，统一使用SQLAlchemy的Connection对象执行所有SQL，让SQLAlchemy统一管理事务：

# 检查数据存在性
existing_row = conn.execute(find_query_string, values).fetchone()

# 插入数据
conn.execute(insert_string, insert_values)
conn.commit()

方法3：开启自动提交（仅用于验证或单操作场景） #

创建引擎时设置自动提交隔离级别，适合单条插入的简单场景（不推荐多操作事务场景）：

engine = create_engine(
    f'mssql+pyodbc://{sql_user}:{sql_password}@{sql_server}/{sql_database}?driver=ODBC+Driver+17+for+SQL+Server',
    isolation_level="AUTOCOMMIT"
)

额外排查项 #

确认查看数据库时未选错数据库实例/架构，比如脚本连接的是TestDB，但SSMS查看的是ProductionDB，或者表属于非dbo架构（如[MySchema].[Timesheets]）。

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