Celery and SQLAlchemy - This result object does not return rows. It has been closed automatically

Question

I have a celery project connected to a MySQL databases. One of the tables is defined like this:

class MyQueues(Base):
    __tablename__ = 'accepted_queues'

    id = sa.Column(sa.Integer, primary_key=True)
    customer = sa.Column(sa.String(length=50), nullable=False)
    accepted = sa.Column(sa.Boolean, default=True, nullable=False)
    denied = sa.Column(sa.Boolean, default=True, nullable=False)

Also, in the settings I have

THREADS = 4

And I am stuck in a function in code.py:

def load_accepted_queues(session, mode=None):

    #make query  
    pool = session.query(MyQueues.customer, MyQueues.accepted, MyQueues.denied)

    #filter conditions    
    if (mode == 'XXX'):
        pool = pool.filter_by(accepted=1)
    elif (mode == 'YYY'):
        pool = pool.filter_by(denied=1)
    elif (mode is None):
        pool = pool.filter(\
            sa.or_(MyQueues.accepted == 1, MyQueues.denied == 1)
            )

   #generate a dictionary with data
   for i in pool: #<---------- line 90 in the error
        l.update({i.customer: {'customer': i.customer, 'accepted': i.accepted, 'denied': i.denied}})

When running this I get an error:

[20130626 115343] Traceback (most recent call last):
  File "/home/me/code/processing/helpers.py", line 129, in wrapper
    ret_value = func(session, *args, **kwargs)
  File "/home/me/code/processing/test.py", line 90, in load_accepted_queues
    for i in pool: #generate a dictionary with data
  File "/home/me/envs/me/local/lib/python2.7/site-packages/sqlalchemy/orm/query.py", line 2341, in instances
    fetch = cursor.fetchall()
  File "/home/me/envs/me/local/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 3205, in fetchall
    l = self.process_rows(self._fetchall_impl())
  File "/home/me/envs/me/local/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 3174, in _fetchall_impl
    self._non_result()
  File "/home/me/envs/me/local/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 3179, in _non_result
    "This result object does not return rows. "
ResourceClosedError: This result object does not return rows. It has been closed automatically

So mainly it is the part

ResourceClosedError: This result object does not return rows. It has been closed automatically

and sometimes also this error:

DBAPIError: (Error) (, AssertionError('Result length not requested length:\nExpected=1. Actual=0. Position: 21. Data Length: 21',)) 'SELECT accepted_queues.customer AS accepted_queues_customer, accepted_queues.accepted AS accepted_queues_accepted, accepted_queues.denied AS accepted_queues_denied \nFROM accepted_queues \nWHERE accepted_queues.accepted = %s OR accepted_queues.denied = %s' (1, 1)

I cannot reproduce the errror properly as it normally happens when processing a lot of data. I tried to change THREADS = 4 to 1 and errors disappeared. Anyway, it is not a solution as I need the number of threads to be kept on 4.

Also, I am confused about the need to use

for i in pool: #<---------- line 90 in the error

or

for i in pool.all(): #<---------- line 90 in the error

and could not find a proper explanation of it.

All together: any advise to skip these difficulties?

Answer 1

For me I got this error when I forgot to write the table calss name for the select function query = select().where(Assessment.created_by == assessment.created_by) so I had only to fix this by adding the class table name I want to get entries from like so:

query = select(Assessment).where(
        Assessment.created_by == assessment.created_by)

Answer 2

I was using an INSERT statment. Adding

RETURNING id

at the end of the query worked for me. As per this issue

That being said it's a pretty weird solution, maybe something fixed in later versions of SQLAlchemy, I am using 1.4.39.

Answer 3

I got the same error while making a query to SQL-Server procedure using SQLAlchemy. In my case, adding SET NOCOUNT ON to the stored procedure fixed the problem.

ALTER PROCEDURE your_procedure_name
AS
BEGIN

    -- SET NOCOUNT ON added to prevent extra result sets from
    -- interfering with SELECT statements.
    SET NOCOUNT ON;

    -- Insert statements for your procedure here
    SELECT *
    FROM your_table_name;

END;

Check out this article for more details

Answer 4

This error occurred for me when I used a variable in Python
and parsed it with an UPDATE statement using pandas pd.read_sql()

Solution:

I simply used mycursor.execute() instead of pd.read_sql()

import mysql.connector and from sqlalchemy import create_engine

Before:

pd.read_sql("UPDATE table SET column = 1 WHERE column = '%s'" % variable, dbConnection)

After:

mycursor.execute("UPDATE table SET column = 1 WHERE column = '%s'" % variable)

Full code:

import mysql.connector
from sqlalchemy import create_engine
import pandas as pd


# Database Connection Setup >
sqlEngine = create_engine('mysql+pymysql://root:root@localhost/db name')
dbConnection = sqlEngine.connect()

db = mysql.connector.connect(
    host="localhost",
    user="root",
    passwd="root",
    database="db name")

mycursor = db.cursor()

variable = "Alex"
mycursor.execute("UPDATE table SET column = 1 WHERE column = '%s'" % variable)

Answer 5

All together: any advise to skip these difficulties?

yes. you absolutely cannot use a Session (or any objects which are associated with that Session), or a Connection, in more than one thread simultaneously, especially with MySQL-Python whose DBAPI connections are very thread-unsafe*. You must organize your application such that each thread deals with it's own, dedicated MySQL-Python connection (and therefore SQLAlchemy Connection/ Session / objects associated with that Session) with no leakage to any other thread.

• Edit: alternatively, you can make use of mutexes to limit access to the Session/Connection/DBAPI connection to just one of those threads at a time, though this is less common because the high degree of locking needed tends to defeat the purpose of using multiple threads in the first place.