एसक्यूएलकेमी

Question

के साथ संग्रहीत प्रक्रियाओं को एसक्यूएल सर्वर के साथ एसक्यूएल सर्वर की संग्रहीत प्रक्रियाओं को कैसे कॉल कर सकता हूं?

Answer 1

इंजन और कनेक्शन में execute() विधि है जिसका आप मनमाने ढंग से एसक्यूएल स्टेटमेंट्स के लिए उपयोग कर सकते हैं, और सत्र भी कर सकते हैं।

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

आप outparam() का उपयोग उत्पादन पैरामीटर बनाना सकते हैं यदि आप (या बाँध मापदंडों isoutparam=True विकल्प के साथ bindparam() का उपयोग के लिए)

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

मान लीजिए कि आपके पास सत्रमेकर() के साथ पहले से सत्र बनाया गया है, तो आप निम्न फ़ंक्शन का उपयोग कर सकते हैं:

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 

इसे कहते वाक्य रचना के साथ काम करता है, तो MySQL काम करना चाहिए। MSSQL कॉल के बजाय EXEC का उपयोग करता है और थोड़ा differennt वाक्यविन्यास, मुझे लगता है। तो इसे सर्वर अज्ञेयवादी बनाना आपके ऊपर है लेकिन बहुत कठिन नहीं होना चाहिए।

Answer 5

SQLAlchemy का उपयोग करके MySQL में संग्रहीत प्रक्रिया को कॉल करने का सबसे आसान तरीका Engine.raw_connection() का callproc विधि का उपयोग कर है। 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()