Python执行Azure SQL存储过程无报错但未修改数据，SSMS/数据工作室执行正常

看起来你遇到了一个很典型的「程序执行无报错但没效果」的问题，这种情况通常是事务、参数、权限或者调用对象不一致导致的，我给你梳理几个排查方向和解决办法：

1. 先确认是不是调用了错误的存储过程 #

你手动执行的是[Construction].[uspProcess]，但Python代码里调用的是SKUK_Lidat.uspProcessLidat——这两个存储过程是不是同一个？有没有可能是schema（比如Construction vs SKUK_Lidat）或者存储过程名称写错了？这是最容易忽略的点，先核对清楚Python里的存储过程全名和你手动执行的完全一致。

2. 检查参数传递是否准确 #

虽然代码里用了命名参数，但还是要确认传入的table_name和batch_id是不是和手动执行的完全一样（比如'Equipment'和26）。可以在Python代码里先打印这两个参数的值，或者在存储过程内部加日志逻辑，比如：

CREATE OR ALTER PROCEDURE SKUK_Lidat.uspProcessLidat
    @TableName NVARCHAR(100),
    @BatchId INT
AS
BEGIN
    -- 加日志到临时表或者专门的日志表
    INSERT INTO ProcessLog (TableName, BatchId, ExecutionTime)
    VALUES (@TableName, @BatchId, GETDATE());
    
    -- 原来的存储过程逻辑...
END

执行完Python脚本后，去查这个日志表，看看存储过程有没有接收到正确的参数，甚至有没有被执行过。

3. 排查事务与自动提交的冲突 #

你的连接字符串和SQL执行都加了autocommit=True，但用SQLAlchemy的session来执行的时候，session的事务管理可能会和这个设置冲突。SQLAlchemy的session默认会把执行逻辑包裹在事务里，即使连接开了autocommit，也可能导致实际没有提交。试试修改代码，去掉execution_options(autocommit=True)，改为手动提交事务，同时打印受影响的行数：

def staging_to_target(table_name, batch_id):
    session = session_factory()  # Create a new session instance
    try:
       sql = text("EXEC SKUK_Lidat.uspProcessLidat @TableName = :table_name, @BatchId = :batch_id")
       result = session.execute(sql, {"table_name": table_name, "batch_id": batch_id})
       session.commit()  # 手动提交事务
       print(f"存储过程执行后受影响行数：{result.rowcount}")
       # 如果rowcount是0，说明存储过程没处理任何数据；如果大于0但数据没变化，再查事务问题
       return True
    except Exception as e:
       session.rollback()  # 出错时回滚事务
       print(f"调用uspProcessLidat出错: {e}")
       return False
    finally:
       session.close()  # 确保session关闭

如果rowcount返回0，那说明存储过程内部的逻辑没有匹配到要迁移的数据，需要去查存储过程的过滤条件；如果rowcount大于0但数据没变化，那就是事务没有被正确提交。

4. 检查执行账号的权限 #

手动执行用的是你的个人账号，而Python用的是托管身份（Managed Identity）或者另一个SQL账号，这个账号有没有足够的权限？比如：

• 有没有读取ConstructionStaging.Equipment表的权限？
• 有没有修改Construction.Equipment表的权限？
• 有没有执行目标存储过程的权限？
  可以在SSMS里用Python的账号登录，手动执行一遍存储过程，看看能不能正常迁移数据。如果不行，就给这个账号加对应的权限。

5. 确认连接的数据库是否正确 #

有时候连接字符串里的数据库名称配置错了，导致Python连接到了另一个数据库，自然看不到数据变化。可以在Python里先执行一个简单的查询，确认当前连接的数据库：

def check_current_db():
    session = session_factory()
    try:
        result = session.execute(text("SELECT DB_NAME() AS CurrentDB"))
        print(f"当前连接的数据库：{result.scalar()}")
    finally:
        session.close()

调用这个函数，看看是不是和你手动执行时用的数据库一致。

备注：内容来源于stack exchange，提问作者Rahul Sharma