Python 使用 sqlAlchemy 存储过程

stored procedures with sqlAlchemy

提问by AKM

How can I call stored procedures of sql server with sqlAlchemy?

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

回答by Steven

Engines and Connections have an execute()method you can use for arbitrary sql statements, and so do Sessions. For example:

引擎和连接有一个execute()可以用于任意 sql 语句的方法，会话也是如此。例如：

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

You can use outparam()to create output parameters if you need to (or for bind parameters use bindparam()with the isoutparam=Trueoption)

outparam()如果需要，您可以使用来创建输出参数（或用于绑定参数bindparam()与isoutparam=True选项一起使用）

回答by Denis Otkidach

Just execute procedure object created with func:

只需执行使用以下方法创建的过程对象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

回答by user1646610

Supposing you already have session created with sessionmaker(), you can use following function:

假设您已经使用 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()

Now you can execute your stored procedure 'MyProc' with parameters simply like that:

现在，您可以使用如下参数执行存储过程“MyProc”：

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

回答by Profpatsch

Out of desperate need for a project of mine, I wrote a function that handles Stored Procedure calls.

出于对我的项目的迫切需要，我编写了一个处理存储过程调用的函数。

Here you go:

干得好：

import sqlalchemy as sql

def execute_db_store_procedure(database, types, sql_store_procedure, *sp_args):
    """ Execute the store procedure and return the response table.

    Attention: No injection checking!!!

    Does work with the CALL syntax as of yet (TODO: other databases).

    Attributes:
        database            -- the database
        types               -- tuple of strings of SQLAlchemy type names.
                               Each type describes the type of the argument
                               with the same number.
                               List: http://docs.sqlalchemy.org/en/rel_0_7/core/types.html
        sql_store_procudure -- string of the stored procedure to be executed
        sp_args             -- arguments passed to the stored procedure
    """
    if not len(types) == len(sp_args):
        raise ValueError("types tuple must be the length of the sp args.")

    # Construch the type list for the given types
    # See
    # http://docs.sqlalchemy.org/en/latest/core/sqlelement.html?highlight=expression.text#sqlalchemy.sql.expression.text
    # sp_args (and their types) are numbered from 0 to len(sp_args)-1
    type_list = [sql.sql.expression.bindparam(
                    str(no), type_=getattr(sql.types, typ)())
                        for no, typ in zip(range(len(types)), types)]

    try:
        # Adapts to the number of arguments given to the function
        sp_call = sql.text("CALL `%s`(%s)" % (
                sql_store_procedure,
                ", ".join([":%s" % n for n in range(len(sp_args))])),
            bindparams=type_list
        )
        #raise ValueError("%s\n%s" % (sp_call, type_list))
        with database.engine.begin() as connection:
            return connection.execute(
                sp_call,
                # Don't do this at home, kids...
                **dict((str(no), arg)
                    for (no, arg) in zip(range(len(sp_args)), sp_args)))
    except sql.exc.DatabaseError:
        raise

It works with the CALL syntax, so MySQL should work as expected. MSSQL uses EXEC instead of call and a little differennt syntax, I guess. So making it server agnostic is up to you but shouldn't be too hard.

它使用 CALL 语法，所以 MySQL 应该按预期工作。我猜，MSSQL 使用 EXEC 而不是调用和一些不同的语法。所以让它与服务器无关取决于你，但不应该太难。

回答by Jofel Bayron

The easiest way to call a stored procedure in MySQL using SQLAlchemy is by using callprocmethod of Engine.raw_connection(). call_procwill require the procedure name and parameters required for the stored procedure being called.

调用使用的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()

回答by Derlin

context: I use flask-sqlalchemy with MySQL and without ORM-mapping. Usually, I use:

上下文：我将flask-sqlalchemy 与MySQL 一起使用，但没有使用ORM 映射。通常，我使用：

# in the init method
_db = SqlAlchemy(app)

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

---

---

Calling stored procedures is not supported out of the box: the callprocis not generic, but specific to the mysql connector.

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

For stored procedures without out params, it is possible to execute a query like

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

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

as usual. Things get more complicated when you have out params...

照常。当您没有参数时，事情会变得更加复杂......

---

---

One way to use out params is to access the underlying connector is through engine.raw_connection(). For example:

使用参数的一种方法是访问底层连接器是通过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>)

This is nice since we are able to access the out parameter, BUT this connection is not managed by the flask session. This means that it won't be committed/aborted as with the other managed queries... (problematic only if your procedure has side-effect).

这很好，因为我们可以访问 out 参数，但是这个连接不是由烧瓶会话管理的。这意味着它不会像其他托管查询一样被提交/中止......（只有在您的过程有副作用时才会出现问题）。

Finally, I ended up doing this:

最后，我最终这样做了：

# 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()

The resultwill be a tuple with one value: the out param. This is not ideal, but the least dangerous: if another query fails during the session, the procedure call will be aborted (rollback) as well.

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

回答by Jerry J

Another workaround:

另一种解决方法：

query = f'call Procedure ("{@param1}", "{@param2}", "{@param3}")'    
sqlEngine = sqlalchemy.create_engine(jdbc)
conn = sqlEngine.connect() 
df = pd.read_sql(query,conn,index_col=None)