gespeicherte Prozeduren mit sqlAlchemy

Question

Wie kann ich gespeicherte Prozeduren von SQL Server mit sqlAlchemy aufrufen?

Answer 1

Engines und Verbindungen haben eine execute() Methode, die Sie für beliebige SQL-Anweisungen verwenden können, und so auch Sessions. Zum Beispiel:

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

können Sie outparam() verwenden, um Ausgabeparameter zu erstellen, wenn Sie müssen (oder binden Parameter bindparam() mit der isoutparam=True Option)

Answer 2

Gerade Prozedur Objekt ausführen mit func erstellt:

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

Angenommen, Sie haben bereits eine Sitzung mit sessionmaker() erstellt, können Sie folgende Funktion verwenden:

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

Jetzt können Sie Ihre gespeicherte Prozedur ‚MyProc‘ mit Parametern ausführen einfach so:

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

Answer 4

Von einer verzweifelten Notwendigkeit für ein Projekt von mir, ich eine Funktion geschrieben, die Prozedur Griffe gespeicherten Anrufe.

Hier gehen Sie:

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 

Es arbeitet mit der CALL-Syntax, so MySQL sollte wie erwartet funktionieren. MSSQL verwendet EXEC anstelle von Aufruf und etwas unterschiedlicher Syntax, denke ich. Es serverunabhängig zu machen, liegt an Ihnen, sollte aber nicht zu schwer sein.

Answer 5

Der einfachste Weg, um eine gespeicherte Prozedur in MySQL mit SQLAlchemy aufzurufen, ist die Methode von Engine.raw_connection(). call_proc erfordert den Prozedurnamen und die Parameter, die für den Aufruf der gespeicherten Prozedur erforderlich sind.

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