使用sqlAlchemy存储过程

Question

如何用sqlAlchemy调用sql server的存储过程？

Answer 1

Engines和Connections有一个execute()可用于任意sql语句的方法,Sessions也是如此.例如:

results = sess.execute('myproc ?, ?', [param1, param2])

您可以使用outparam()创建的输出参数,如果你需要(或绑定参数使用bindparam()与isoutparam=True选项)

Answer 2

只需执行创建的过程对象func:

from sqlalchemy import create_engine, func
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite://', echo=True)
print engine.execute(func.upper('abc')).scalar() # Using engine
session = sessionmaker(bind=engine)()
print session.execute(func.upper('abc')).scalar() # Using session

Answer 3

context：我在MySQL和不带ORM映射的情况下使用flask-sqlalchemy。通常，我使用：

# in the init method
_db = SqlAlchemy(app)

#... somewhere in my code ...
_db.session.execute(query)

开箱即用的存储过程不支持：callproc不是通用的，而是特定于mysql连接器的。

对于没有参数的存储过程，可以执行如下查询

_db.session.execute(sqlalchemy.text("CALL my_proc(:param)"), param='something')

照常。如果没有参数，事情会变得更加复杂...

用完参数的一种方法是通过访问底层连接器engine.raw_connection()。例如：

conn = _db.engine.raw_connection()
# do the call. The actual parameter does not matter, could be ['lala'] as well
results = conn.cursor().callproc('my_proc_with_one_out_param', [0])
conn.close()   # commit
print(results) # will print (<out param result>)

很好，因为我们能够访问out参数，但是该连接不是由flask会话管理的。这意味着它不会像其他托管查询一样被提交/中止（仅当您的过程有副作用时才有问题）。

最后，我最终这样做：

# do the call and store the result in a local mysql variabl
# the name does not matter, as long as it is prefixed by @
_db.session.execute('CALL my_proc_with_one_out_param(@out)')
# do another query to get back the result
result = _db.session.execute('SELECT @out').fetchone()

该result会是一个值的元组：在出PARAM。这不是理想的方法，但是危险最小：如果在会话期间另一个查询失败，则过程调用也将被中止（回滚）。

Answer 4

假设您已经使用sessionmaker()创建了会话,则可以使用以下函数:

def exec_procedure(session, proc_name, params):
    sql_params = ",".join(["@{0}={1}".format(name, value) for name, value in params.items()])
    sql_string = """
        DECLARE @return_value int;
        EXEC    @return_value = [dbo].[{proc_name}] {params};
        SELECT 'Return Value' = @return_value;
    """.format(proc_name=proc_name, params=sql_params)

    return session.execute(sql_string).fetchall()

现在,您可以使用以下参数执行存储过程"MyProc":

params = {
    'Foo': foo_value,
    'Bar': bar_value
}
exec_procedure(session, 'MyProc', params)

Answer 5

调用使用的SQLAlchemy MySQL中的存储过程的最简单的方法是通过使用callproc的方法Engine.raw_connection()。call_proc将需要被调用的存储过程所需的过程名称和参数。

def call_procedure(function_name, params):
       connection = cloudsql.Engine.raw_connection()
       try:
           cursor = connection.cursor()
           cursor.callproc(function_name, params)
           results = list(cursor.fetchall())
           cursor.close()
           connection.commit()
           return results
       finally:
           connection.close()