Stored Procedures With Sqlalchemy

Calling MSSQL stored procedure from SqlAlchemy

I remember this giving me grief before too. From memory either session.execute() or connection.execute() worked for me. There's also a callproc() method, which is probably the right way to go. http://docs.sqlalchemy.org/en/latest/core/connections.html

Also, I've had issues in the past with MSSQL which seemed to be due to something asynchronous happening where the method was returning before the procedure was finished, which resulted in unpredictable results on the database. I found that putting a time.sleep(1) (or whatever the appropriate number is) right after the call fixed this.

How do I execute a MySQL stored procedure in a sqlalchemy scoped session to return a single result set of data (for flask web app)?

SQLAlchemy doesn't, as far as I know, support calling stored procedures directly. The docs recommend using the raw DB-API connection's callproc method.

The connection can be accessed from the engine; it can also be accessed via a session, but this is still goes through the engine.

The example code below shows both methods. Note that the method of accessing the result of calling the procedure may vary depending on the parameters and the connector used - see this answer for some examples.

import mysql.connector

import sqlalchemy as sa
from sqlalchemy import orm

# Setup the database

DATA = [(1, 2, 3), (4, 5, 6), (7, 8, 9)]

DDL1 = """\
CREATE TABLE IF NOT EXISTS test_table (
  id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
  a INT,
  b INT,
  c INT)
"""

DDL2 = """\
CREATE PROCEDURE IF NOT EXISTS test_procedure (IN p1 INT)
  BEGIN
    SELECT a, b, c FROM test_table
    WHERE a > p1;
  END
"""

DML1 = """DELETE FROM test_table"""

DML2 = """INSERT INTO test_table (a, b, c) VALUES (%s, %s, %s)"""

CALL1 = """CALL test_procedure(:param)"""

conn = mysql.connector.connect(database='test')
cur = conn.cursor()
cur.execute(DDL1)
cur.execute(DDL2)
cur.execute(DML1)
for row in DATA:
    cur.execute(DML2, row)
conn.commit()
conn.close()


# Call the procedure

engine = sa.create_engine('mysql+mysqlconnector:///test')
Session = orm.scoped_session(orm.sessionmaker(autocommit=False, autoflush=False, bind=engine))
session = Session()

raw_conn = session.connection().engine.raw_connection()
cur = raw_conn.cursor()
cur.callproc('test_procedure', [1])
print('Using session')
for result in cur.stored_results():
    print(result.fetchall())
Session.remove()

print('Using engine directly')
raw_conn = engine.raw_connection()
cur = raw_conn.cursor()
cur.callproc('test_procedure', [1])
for result in cur.stored_results():
    print(result.fetchall())

One final observation: it seems to me that the raw connection accessed via the session is outside of the session's transaction context, so changes made using the session might not be visible to changes made using the connection, depending on transaction isolation settings.

Launch SQL stored procedures from python with sqlalchemy?

Finally solved my problem with the following function :

def execute_stored_procedure(engine, procedure_name):
    res = {}
    connection = engine.raw_connection()
    try:
        cursor = connection.cursor()
        cursor.execute("EXEC "+procedure_name)
        cursor.close()
        connection.commit()
        res['status'] = 'OK'
    except Exception as e:
        res['status'] = 'ERROR'
        res['error'] = e
    finally:
        connection.close() 
    return res

why the stored procedure called from sqlalchemy is not working but calling from workbench is working?

I haven't called stored procs from SQLAlchemy, but it seems possible that this could be within a transaction because you're using the session. Perhaps calling db.session.commit() at the end would help?

If that fails, SQLAlchemy calls out calling stored procs here. Perhaps try their method of using callproc. Adapting to your use-case, something like:

connection = db.session.connection()
try:
    cursor = connection.cursor()
    cursor.callproc("escalatelobalarm", [clientid])
    results = list(cursor.fetchall())
    cursor.close()
    connection.commit()
finally:
    connection.close()

How correctly execute a MSSQL stored procedure with parameters in Python

Write your SQL command text using the "named" paramstyle, wrap it in a SQLAlchemy text() object, and pass the parameter values as a dict:

import pandas as pd
import sqlalchemy as sa

connection_uri = "mssql+pyodbc://@mssqlLocal64"
engine = sa.create_engine(connection_uri)

# SQL command text using "named" paramstyle
sql = """
SET NOCOUNT ON;
SET ARITHABORT ON;
EXEC dbo.breakfast @name = :name_param, @food = :food_param;
"""
# parameter values
param_values = {"name_param": "Gord", "food_param": "bacon"}
# execute query wrapped in SQLAlchemy text() object
df = pd.read_sql_query(sa.text(sql), engine, params=param_values)

print(df)
"""
                           column1
0  Gord likes bacon for breakfast.
"""

Python sqlalchemy and mySQL stored procedure always returns 0 (out param only)

With the help of this answer I found the following solution that worked for me.

a) Working solution using engine.raw_connection() and cursor.callproc:

def call_procedure(engine, function_name):
    connection = engine.raw_connection()
    try:
        cursor = connection.cursor()
        cursor.callproc(function_name, [0])
        cursor.execute(f"""SELECT @_{function_name}_0""")
        results = cursor.fetchone() ## returns a tuple e.g. (285,)
        rows_affected = results[0]
        cursor.close()
        connection.commit()
        logger.info(f"Running procedure {function_name} success!")
        return rows_affected
    except Exception as e:
        logger.error(f"Running procedure {function_name} failed!")
        logger.exception(e)
        return None
    finally:
        connection.close()

And with this answer I found this solution also:

b) Instead of using a raw connection, this worked as well:

def call_procedure(engine, function_name, params=None):
    try:
        with engine.begin() as db_conn:
            db_conn.execute(f"""CALL {function_name}(@out)""")
            results = db_conn.execute('SELECT @out').fetchone() ## returns a tuple e.g. (285,)
            rows_affected = results[0]
        logger.debug(f"Running procedure {function_name} success!")
        return rows_affected
    except Exception as e:
        logger.error(f"Running procedure {function_name} failed!")
        logger.exception(e)
        return None
    finally:
        if db_conn: db_conn.close()

If there are any advantages or drawbacks of using one of these methods over the other, please let me know in a comment.

How to call stored procedure with SQLAlchemy that requires a user-defined-type Table parameter

There is a driver that really supports TVPs: Pytds. It's not officially supported, but there's a 3rd party dialect implementation for it: sqlalchemy-pytds. Using them you could call your stored procedure like so:

In [1]: engine.execute(DDL("CREATE TYPE [dbo].[StringTable] AS TABLE([strValue] [nvarchar](max) NULL)"))
Out[1]: <sqlalchemy.engine.result.ResultProxy at 0x7f235809ae48>

In [2]: engine.execute(DDL("CREATE PROC test_proc (@pArg [StringTable] READONLY) AS BEGIN SELECT * FROM @pArg END"))
Out[2]: <sqlalchemy.engine.result.ResultProxy at 0x7f2358027b70>

In [3]: arg = ['Name One', 'Name Two']

In [4]: import pytds

In [5]: tvp = pytds.TableValuedParam(type_name='StringTable',
   ...:                              rows=((x,) for x in arg))

In [6]: engine.execute('EXEC test_proc %s', (tvp,))
Out[6]: <sqlalchemy.engine.result.ResultProxy at 0x7f294e699e10>

In [7]: _.fetchall()
Out[7]: [('Name One',), ('Name Two',)]

This way you can pass potentially large amounts of data as params:

In [21]: tvp = pytds.TableValuedParam(type_name='StringTable',
    ...:                              rows=((str(x),) for x in range(100000)))

In [22]: engine.execute('EXEC test_proc %s', (tvp,))
Out[22]: <sqlalchemy.engine.result.ResultProxy at 0x7f294c6e9f98>

In [23]: _.fetchall()[-1]
Out[23]: ('99999',)

If on the other hand you're using a driver that does not support TVPs, you could declare a table variable, insert the values, and pass that as the argument to your procedure:

In [12]: engine.execute(
    ...:     """
    ...:     DECLARE @pArg AS [StringTable];
    ...:     INSERT INTO @pArg VALUES {placeholders};
    ...:     EXEC test_proc @pArg;
    ...:     """.format(placeholders=",".join(["(%s)"] * len(arg))),
    ...:     tuple(arg))
    ...:     
Out[12]: <sqlalchemy.engine.result.ResultProxy at 0x7f23580f2908>

In [15]: _.fetchall()
Out[15]: [('Name One',), ('Name Two',)]

Note that you cannot use any executemany methods, or you'll end up calling the procedure for each table value separately. That is why the placeholders are constructed manually and the table values passed as individual arguments. Care must be taken not to format any arguments directly in to the query, but the correct amount of placeholders for the DB-API instead. Row values are limited to a maximum of 1000.

It'd of course be nice, if the underlying DB-API driver provided proper support for table valued parameters, but at least I could not find a way for pymssql, which uses FreeTDS. A reference to TVPs on the mailing list makes it clear that they're not supported. The situation is not much better for PyODBC.

_{Disclaimer: I've not really used MS SQL Server before.}

SQLAlchemy stored procedure is not working in Postgres

As noted in a comment to the question, psycopg2 does not like the commit; in the SP. If you remove that then this will work with SQLAlchemy:

import sqlalchemy as sa
from sqlalchemy.orm import Session

engine = sa.create_engine("postgresql://scott:tiger@192.168.0.199/test")

with Session(engine) as session, session.begin():
    session.execute(
        sa.text("CALL addcity (:param1, :param2)"),
        {"param1": "One", "param2": "Two"},
    )
# automatically commits when the context manager exits (assuming no errors)

---

Related Topics