SQLAlchemy and Stored Procedures

I’m so excited that I solved this problem, I had to blog about it.  I couldn’t find anything online to address this specific problem: Getting stored procedure results in key-value format using SQLAlchemy.

I have a stored procedure that does a SELECT query, and requires a few parameters.  Let’s call my stored procedure get_my_data. And it requires three params, userid, url, and date.

Looking at the SQLAlchemy docs, I found this page regarding stored procedures. So I started there.

# My function has a db session, so I had to get the cursor from there
connection = session.connection()
# In order to get the cursor, you must get the connection's connection, which seems insane, but let's go with it.
dbapi_conn = connection.connection
# Finally get the cursor and call the stored procedure, as recommended by the SQLAlchemy docs.
cursor = dbapi_conn.cursor()
cursor.callproc('get_my_data', [userid, url, date])
# Get a list of rows, without the column names
results = cursor.fetchall()
# Clean everything up
cursor.close()
dbapi_conn.close()
# Return the db rows
return results

The problem was with the format of the results. I got a list, and each item in the list was another list of the values. The column names were nowhere to be found. A list of values without keys wasn’t going to help the consumers of this endpoint! So at first I tried to iterate over the values in each row, and create a dict. This was based on the index of the value in the row, which I soon found out wasn’t maintainable if the columns returned changed. Plus, it didn’t feel right. I knew there must be some way to get the column names out of the query without hard-coding which column was at which index.

After some more searching, I found that the cursor has a description property. I could use that to find the column names. Add some Python to the mix, and I had a my list of rows. Each row is now a dict with the correct column name as the key.

# My function has a SQLAlchemy session, so I had to get the cursor from there
connection = session.connection()
# In order to get the cursor, you must get the connection's connection, which seems insane, but let's go with it.
dbapi_conn = connection.connection
# Finally get the cursor and call the stored procedure, as recommended by the SQLAlchemy docs.
cursor = dbapi_conn.cursor()
cursor.callproc('get_my_data', [userid, url, date])
# Get a list of rows, without the column names
results = cursor.fetchall()
# Get the column names for the results
indices = [item[0] for item in cursor.description]
# Create a list of mapped column names to results
results_list = [dict(zip(indices, result)) for result in results]
# Clean everything up
cursor.close()
dbapi_conn.close()
# Return the db rows
return results_list