español - sqlalchemy python

procedimientos almacenados con sqlAlchemy

Question

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

Answer 1

Debido a la necesidad desesperada de un proyecto mío, escribí una función que maneja las llamadas al Procedimiento Almacenado.

Aqui tienes:

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

Funciona con la sintaxis CALL, por lo que MySQL debería funcionar como se esperaba. 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 2

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

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

Answer 3

Simplemente ejecuta el objeto de procedimiento 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 4

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 puede ejecutar su procedimiento almacenado ''MyProc'' con parámetros simplemente así:

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

Answer 5

contexto : uso flask-sqlalchemy con MySQL y sin ORM-mapping. Por lo general, uso:

# in the init method
_db = SqlAlchemy(app)
#... somewhere in my code ...
_db.session.execute(query)

La llamada a procedimientos almacenados no se admite de callproc : el callproc no es genérico, sino específico del conector mysql.

Para procedimientos almacenados sin parámetros , es posible ejecutar una consulta como

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

como siempre. Las cosas se vuelven más complicadas cuando tienes params ...

Una forma de utilizar params es acceder al conector subyacente a través de engine.raw_connection() . Por ejemplo:

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

Esto es bueno, ya que podemos acceder al parámetro out, PERO esta conexión no está gestionada por la sesión del matraz . Esto significa que no se comprometerá / abortará como con las otras consultas administradas ... (problemática solo si su procedimiento tiene efecto secundario).

Finalmente, terminé haciendo esto:

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

El result será una tupla con un valor: el param externo. Esto no es ideal, pero es el menos peligroso: si otra consulta falla durante la sesión, la llamada al procedimiento también se abortará (reversión).

Answer 6

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 usar outparam() para crear parámetros de salida si lo necesita (o para los parámetros de enlace use bindparam() con la opción isoutparam=True )