当你想在你的代码中找到一个错误时,这很难;当你认为你的代码是不会有错误时,这就更难了。

使用sqlalchemy的python存储过程

admin 75℃
如何用sqlalchemy调用sql server的存储过程?

发动机和连接有 execute() 方法可以用于任意SQL语句,会话也可以。例如:

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

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

只执行用创建的过程对象 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

如果您已经使用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)

上下文:我在mysql中使用flask sqlalchemy,而没有orm映射。通常,我使用:

# in the init method
_db = SqlAlchemy(app)

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

不支持立即调用存储过程: callproc 不是泛型的,而是特定于mysql连接器的。

对于没有out params的存储过程,可以执行类似于

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

像往常一样。如果没有参数,事情会变得更复杂…


使用out params的一种方法是通过 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 将是具有一个值的元组:out param。这并不理想,但最危险的是:如果在会话期间另一个查询失败,过程调用也将中止(回滚)。

转载请注明:我的代码 » 使用sqlalchemy的python存储过程