Python调用MSSQL存储过程遇TypeError：Connection.execute()参数异常

尝试用Python脚本调用MSSQL的存储过程Sp_Test向表中插入数据，存储过程代码如下：

SET ANSI_NULLS ON GO
SET QUOTED_IDENTIFIER ON GO
CREATE PROCEDURE [dbo].[Sp_Test] 
( 
   @ProcessId INT
  ,@CreatedByUserName VARCHAR(12)
  ,@CreatedByMachineName VARCHAR(12)
  --,@CreatedTime DATETIME
  ,@NumberOfDataRows INT  
  ,@DataSource VARCHAR(300)
  --,@RefId VARCHAR(8)
  ,@Postpone DATETIME2
  ,@Deadline DATETIME2
  ,@NewBatchId INT OUT 
) 
AS  
BEGIN    
INSERT INTO Test(ProcessId, CreatedByUserName,
CreatedByMachineName, CreatedTime, Postpone, Deadline,
NumberOfDataRows, DataSource, RefId)   
VALUES(
 @ProcessId,
 @CreatedByUserName, 
 @CreatedByMachineName, 
 GETDATE(), 
 NULL, 
 NULL,
 @NumberOfDataRows, 
 @DataSource, 
 CONVERT(VARCHAR(9),
 CRYPT_GEN_RANDOM(4), 2)
)
SET @NewBatchId = SCOPE_IDENTITY();
RETURN 0; 
END 
GO

最初的Python代码片段：

with engine.connect() as conn:
    conn.execute(text("CALL Sp_Test(:ProcessId, :CreatedByUserName, :CreatedByMachineName, :NumberOfDataRows, :DataSource)"), ProcessId=process_code, CreatedByUserName=username,CreatedByMachineName=hostname,NumberOfDataRows=row_count,DataSource='Shared')

运行时触发错误：

TypeError: Connection.execute() got an unexpected keyword argument 'ProcessId'

改用字典传参后仍报相同错误：

params = {'ProcessId': process_code,'CreatedByUserName': username,'CreatedByMachineName': hostname,'NumberOfDataRows': row_count,'DataSource': 'Shared'} 
with engine.connect() as conn: 
    conn.execute(text("CALL Sp_Test(:ProcessId, :CreatedByUserName, :CreatedByMachineName, :NumberOfDataRows, :DataSource)"),**params)

1. 修正参数传递格式 #

SQLAlchemy的Connection.execute()方法不支持直接传入关键字参数，必须将参数作为第二个位置参数传入，可选择字典或列表格式：

字典传参（按名称匹配） #

params = {
    'ProcessId': process_code,
    'CreatedByUserName': username,
    'CreatedByMachineName': hostname,
    'NumberOfDataRows': row_count,
    'DataSource': 'Shared',
    'Postpone': None,
    'Deadline': None
} 
with engine.connect() as conn: 
    conn.execute(text("CALL Sp_Test(:ProcessId, :CreatedByUserName, :CreatedByMachineName, :NumberOfDataRows, :DataSource, :Postpone, :Deadline, @NewBatchId OUT)"), params)

列表传参（按位置匹配） #

params = [
    process_code,
    username,
    hostname,
    row_count,
    'Shared',
    None,
    None
]
with engine.connect() as conn:
    conn.execute(text("CALL Sp_Test(?, ?, ?, ?, ?, ?, ?, @NewBatchId OUT)"), params)

2. 补全存储过程的完整参数列表 #

你的存储过程定义了8个参数（含OUT参数），但最初调用只传入了5个，遗漏了@Postpone、@Deadline和OUT参数@NewBatchId。即使存储过程内部将前两个参数设为NULL，也需要显式传入对应值，否则会触发参数不匹配问题。

3. 获取OUT参数返回值（可选） #

如果需要获取存储过程返回的@NewBatchId，可以通过声明变量并查询的方式实现：

from sqlalchemy import text

with engine.connect() as conn:
    new_batch_id = conn.execute(
        text("DECLARE @NewBatchId INT; EXEC Sp_Test :ProcessId, :CreatedByUserName, :CreatedByMachineName, :NumberOfDataRows, :DataSource, :Postpone, :Deadline, @NewBatchId OUT; SELECT @NewBatchId AS NewBatchId"),
        {
            'ProcessId': process_code,
            'CreatedByUserName': username,
            'CreatedByMachineName': hostname,
            'NumberOfDataRows': row_count,
            'DataSource': 'Shared',
            'Postpone': None,
            'Deadline': None
        }
    ).scalar()

print(f"生成的BatchId: {new_batch_id}")

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