用sqlAlchemy存儲過程

Question

如何用sqlAlchemy調用sql server存儲過程？

Answer 1

引擎和連接有一個​​方法可以用於任意的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

假設您已經使用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 4

出於迫切需要爲我的項目，我寫了一個處理存儲過程調用的函數。

在這裏你去：

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 

它的工作原理與CALL語法，因此MySQL應該按預期工作。我想，MSSQL使用EXEC而不是調用和一些不同的語法。因此，使其成爲服務器不可知論取決於你，但不應該太難。

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