CODEWITHSUNDEEP
pythonunit-testingflasksqlalchemycelery
deBrice
deBrice

Reputation: 485

Connection is closed when a SQLAlchemy event triggers a Celery task

When one of my unit tests deletes a SQLAlchemy object, the object triggers an after_delete event which triggers a Celery task to delete a file from the drive.

The task is CELERY_ALWAYS_EAGER = True when testing.

gist to reproduce the issue easily

The example has two tests. One triggers the task in the event, the other outside the event. Only the one in the event closes the connection.

To quickly reproduce the error you can run:

git clone https://gist.github.com/5762792fc1d628843697.git
cd 5762792fc1d628843697
virtualenv venv
. venv/bin/activate
pip install -r requirements.txt
python test.py

The stack:

$     python test.py
E
======================================================================
ERROR: test_delete_task (__main__.CeleryTestCase)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "test.py", line 73, in test_delete_task
    db.session.commit()
  File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/scoping.py", line 150, in do
    return getattr(self.registry(), name)(*args, **kwargs)
  File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 776, in commit
    self.transaction.commit()
  File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 377, in commit
    self._prepare_impl()
  File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 357, in _prepare_impl
    self.session.flush()
  File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 1919, in flush
    self._flush(objects)
  File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 2037, in _flush
    transaction.rollback(_capture_exception=True)
  File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/util/langhelpers.py", line 63, in __exit__
    compat.reraise(type_, value, traceback)
  File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 2037, in _flush
    transaction.rollback(_capture_exception=True)
  File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 393, in rollback
    self._assert_active(prepared_ok=True, rollback_ok=True)
  File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 223, in _assert_active
    raise sa_exc.ResourceClosedError(closed_msg)
ResourceClosedError: This transaction is closed

----------------------------------------------------------------------
Ran 1 test in 0.014s

FAILED (errors=1)

Upvotes: 7

Views: 4909

Answers (3)

jbasko
jbasko

Reputation: 7330

Similar to the answer suggested by deBrice, but using the approach similar to Rachel.

class ContextTask(TaskBase):
    abstract = True

    def __call__(self, *args, **kwargs):
        import flask
        # tests will be run in unittest app context
        if flask.current_app:
            return TaskBase.__call__(self, *args, **kwargs)
        else:
            # actual workers need to enter worker app context 
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

Upvotes: 2

deBrice
deBrice

Reputation: 485

Ask, the creator of celery, suggested that solution on github

from celery import signals

def make_celery(app):
     ...

     @signals.task_prerun.connect
     def add_task_flask_context(sender, **kwargs):
         if not sender.request.is_eager:
            sender.request.flask_context = app.app_context().__enter__()

    @signals.task_postrun.connect
    def cleanup_task_flask_context(sender, **kwargs):
       flask_context = getattr(sender.request, 'flask_context', None)
       if flask_context is not None:
           flask_context.__exit__(None, None, None)

Upvotes: 0

Rachel Sanders
Rachel Sanders

Reputation: 5874

I think I found the problem - it's in how you set up your Celery task. If you remove the app context call from your celery setup, everything runs fine:

class ContextTask(TaskBase):
    abstract = True

    def __call__(self, *args, **kwargs):
        # deleted --> with app.app_context():
        return TaskBase.__call__(self, *args, **kwargs)

There's a big warning in the SQLAlchemy docs about never modifying the session during after_delete events: http://docs.sqlalchemy.org/en/latest/orm/events.html#sqlalchemy.orm.events.MapperEvents.after_delete

So I suspect the with app.app_context(): is being called during the delete, trying to attach to and/or modify the session that Flask-SQLAlchemy stores in the app object, and therefore the whole thing is bombing.

Flask-SQlAlchemy does a lot of magic behind the scenes for you, but you can bypass this and use SQLAlchemy directly. If you need to talk to the database during the delete event, you can create a new session to the db:

@celery.task()
def my_task():
    # obviously here I create a new object
    session = db.create_scoped_session()
    session.add(User(id=13, value="random string"))
    session.commit()
    return

But it sounds like you don't need this, you're just trying to delete an image path. In that case, I would just change your task so it takes a path:

# instance will call the task
@event.listens_for(User, "after_delete")
def after_delete(mapper, connection, target):
    my_task.delay(target.value)

@celery.task()
def my_task(image_path):
    os.remove(image_path)

Hopefully that's helpful - let me know if any of that doesn't work for you. Thanks for the very detailed setup, it really helped in debugging.

Upvotes: 8

Related Questions