procedimientos almacenados con sqlAlchemy

Question

¿Cómo puedo llamar a los procedimientos almacenados del servidor sql con sqlAlchemy?

Answer 1

Los motores y las conexiones tienen un método execute() que puede usar para las sentencias sql arbitrarias, y también lo hacen las sesiones. Por ejemplo:

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

Puede utilizar outparam() para crear parámetros de salida si es necesario (o para los parámetros, use bind bindparam() con la opción isoutparam=True)

Answer 2

sólo hay que ejecutar procedimiento objeto creado 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

Suponiendo que ya tiene una sesión creada con sessionmaker(), puede usar la siguiente función:

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

Ahora usted puede ejecutar su procedimiento 'MyProc' almacenado con parámetros simplemente por el estilo:

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

Answer 4

Fuera de desesperada necesidad de un proyecto mío, escribí una función que se encarga de procedimiento almacenado llama.

Aquí van:

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 

Se trabaja con la sintaxis de llamada, por lo MySQL debería funcionar como se espera. MSSQL usa EXEC en lugar de llamada y una pequeña sintaxis, supongo. Así que hacerlo independiente del servidor depende de usted, pero no debería ser demasiado difícil.

Answer 5

La forma más fácil de llamar a un procedimiento almacenado en MySQL utilizando SQLAlchemy es mediante el método callproc de Engine.raw_connection(). call_proc requerirá el nombre del procedimiento y los parámetros necesarios para el procedimiento almacenado que se llama.

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