SQLAlchemy returns stale data in celery tasks

Question

My current setup include

Flask, Flask-SQLAlchemy, celery

Now the problem which I am running into is sometimes database queries inside celery tasks gives stale data i.e If I ask for last record from table I get second last while the last entry was inserted just 10-15 minutes before doing the query and sometimes I see exceptions like these.

OperationalError("(OperationalError) (2006, \'MySQL server has gone away\')",)'

Following is trackback

  File "/home/sys_user/repo/my_app/app/tasks/reminders.py", line 63, in run
    config = Reminder.query.filter_by(id=reminder_id).first()
  File "/home/sys_user/Envs/flask/local/lib/python2.7/site-packages/sqlalchemy/orm/query.py", line 2341, in first
    ret = list(self[0:1])
  File "/home/sys_user/Envs/flask/local/lib/python2.7/site-packages/sqlalchemy/orm/query.py", line 2208, in __getitem__
    return list(res)
  File "/home/sys_user/Envs/flask/local/lib/python2.7/site-packages/sqlalchemy/orm/query.py", line 2412, in __iter__
    return self._execute_and_instances(context)
  File "/home/sys_user/Envs/flask/local/lib/python2.7/site-packages/sqlalchemy/orm/query.py", line 2427, in _execute_and_instances
    result = conn.execute(querycontext.statement, self._params)
  File "/home/sys_user/Envs/flask/local/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 729, in execute
    return meth(self, multiparams, params)
  File "/home/sys_user/Envs/flask/local/lib/python2.7/site-packages/sqlalchemy/sql/elements.py", line 321, in _execute_on_connection
    return connection._execute_clauseelement(self, multiparams, params)
  File "/home/sys_user/Envs/flask/local/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 826, in _execute_clauseelement
    compiled_sql, distilled_params
  File "/home/sys_user/Envs/flask/local/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 958, in _execute_context
    context)
  File "/home/sys_user/Envs/flask/local/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 1160, in _handle_dbapi_exception
    exc_info
  File "/home/sys_user/Envs/flask/local/lib/python2.7/site-packages/sqlalchemy/util/compat.py", line 199, in raise_from_cause
    reraise(type(exception), exception, tb=exc_tb)
  File "/home/sys_user/Envs/flask/local/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 951, in _execute_context
    context)
  File "/home/sys_user/Envs/flask/local/lib/python2.7/site-packages/sqlalchemy/engine/default.py", line 436, in do_execute
    cursor.execute(statement, parameters)
  File "/home/sys_user/Envs/flask/local/lib/python2.7/site-packages/MySQLdb/cursors.py", line 205, in execute
    self.errorhandler(self, exc, value)
  File "/home/sys_user/Envs/flask/local/lib/python2.7/site-packages/MySQLdb/connections.py", line 36, in defaulterrorhandler
    raise errorclass, errorvalue

Following is my celery_app.py

from app import app as flask_app, db
from celery.signals import worker_process_init

def make_celery(app=None):
    celery = Celery()
    celery.conf.update(flask_app.config)
    celery.config_from_object(celeryconfig)

    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with flask_app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    db.init_app(flask_app)
    celery.Task = ContextTask
    return celery


celery_instance = make_celery()


@worker_process_init.connect
def celery_worker_init_db(**_):
    db.init_app(flask_app)

Some tasks which are defined.

class ReminderTask(Task):
    ignore_result = True

    def run(self, data):
        a = Reminder.query.filter_by(id=int(data['reminder_id'])).first()
        a.send_reminders()

Answer 1

This same question got answered at sqlalchemy google group

Based on that I ended up implementing the solution which ask mentions here. I haven't seen these issues since then.