使用SQLAlchemy执行SQL Server无参存储过程时遇参数错误

错误根源 #

你遇到的“mapping or sequence expected for parameters”错误，是因为调用session.execute()时错误地将engine作为第二个参数传入。execute()方法的第二个参数用于传递SQL语句所需的参数绑定值（要求是字典或序列类型），但你的存储过程不需要参数，完全不需要传递这个参数。

修正后的代码 #

把两处execute调用里的engine参数去掉即可：

def mssqlDataPrep():
    try:
        engine = create_engine('mssql+pyodbc://@' + srvr + '/' + db + '?trusted_connection=yes&driver=ODBC+Driver+13+for+SQL+Server')

        Session = scoped_session(sessionmaker(bind=engine))
        s = Session()
        
        src_tables = s.execute("""select t.name as table_name from sys.tables t where t.name in ('UPrices') union select t.name as table_name from sys.tables t where t.name in ('ExtractViewFromPrices')""")

        for tbl in src_tables:
            if str(tbl[0]) == 'ExtractViewFromPrices':
                populateFromSrcVwQry = '''exec stg.PopulateExtractViewFromPrices'''
                exec_sproc_extract = s.execute(populateFromSrcVwQry)
            else:
                populateUQry = '''exec stg.PopulateUPrices''' 
                exec_sproc_u = s.execute(populateUQry)
        
        # 存储过程涉及数据修改时，记得提交事务
        s.commit()
    except Exception as e:
        # 出错时回滚事务，避免数据不一致
        s.rollback()
        print("Data prep error: " + str(e))
    finally:
        # 关闭会话，释放数据库连接资源
        s.close()

额外优化提示 #

• 因为存储过程没有参数，无需使用f-string，直接用普通字符串更简洁。
• 建议添加事务提交/回滚逻辑，避免数据异常；最后关闭会话释放资源。
• 也可以用SQLAlchemy的text()对象包裹SQL语句，更符合ORM规范：

  from sqlalchemy import text
  
  # 调用示例
  s.execute(text('exec stg.PopulateExtractViewFromPrices'))
  

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