stored procedure con sqlAlchemy

Question

Come posso chiamare stored procedure di sql server con sqlAlchemy?

Answer 1

I motori e i collegamenti hanno un metodo execute() che è possibile utilizzare per istruzioni arbitrarie di SQL, e quindi anche Sessioni. Per esempio:

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

È possibile utilizzare outparam() per creare parametri di output se è necessario (o per i parametri legano utilizzano bindparam() con l'opzione isoutparam=True)

Answer 2

Basta eseguire oggetto procedura creata con 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

Supponendo di aver già creato una sessione con sessionmaker(), è possibile utilizzare la seguente funzione:

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

Ora è possibile eseguire il procedimento 'MyProc' memorizzato con i parametri semplicemente così:

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

Answer 4

Fuori disperato bisogno di un mio progetto, ho scritto una funzione che gestisce Stored Procedure Calls.

Qui si va:

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 

Funziona con la sintassi CALL, quindi MySQL dovrebbe funzionare come previsto. MSSQL usa EXEC al posto della chiamata e una sintassi leggermente diversa, suppongo. Quindi renderlo agnostico da server dipende da te, ma non dovrebbe essere troppo difficile.

Answer 5

Il modo più semplice per chiamare una stored procedure in MySQL utilizzando SQLAlchemy è utilizzando il metodo di Engine.raw_connection(). call_proc richiederà il nome della procedura e i parametri richiesti per la chiamata alla procedura memorizzata.

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