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_taskdb.session.commit()File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/scoping.py", line 150, in doreturn 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 commitself.transaction.commit()File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 377, in commitself._prepare_impl()File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 357, in _prepare_implself.session.flush()File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 1919, in flushself._flush(objects)File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 2037, in _flushtransaction.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 _flushtransaction.rollback(_capture_exception=True)File "/home/brice/Code/5762792fc1d628843697/venv/local/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 393, in rollbackself._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_activeraise sa_exc.ResourceClosedError(closed_msg)
ResourceClosedError: This transaction is closed----------------------------------------------------------------------
Ran 1 test in 0.014sFAILED (errors=1)

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 = Truedef __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 objectsession = 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.