python - 当 SQLAlchemy 事件触发 Celery 任务时连接关闭

Question

当我的一个单元测试删除一个 SQLAlchemy 对象时，该对象会触发一个 after_delete 事件，该事件会触发一个 Celery 任务以从驱动器中删除一个文件。

测试时任务为CELERY_ALWAYS_EAGER = True。

gist to reproduce the issue easily

该示例有两个测试。一个在事件中触发任务，另一个在事件外触发。只有事件中的那个关闭连接。

要快速重现错误，您可以运行:

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

堆栈:

$     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)

Answer 1

我想我找到了问题所在 - 问题出在您设置 Celery 任务的方式上。如果您从 celery 设置中删除应用程序上下文调用，一切运行正常:

class ContextTask(TaskBase):
    abstract = True

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

SQLAlchemy 文档中有一个关于在 after_delete 事件期间从不修改 session 的重要警告:http://docs.sqlalchemy.org/en/latest/orm/events.html#sqlalchemy.orm.events.MapperEvents.after_delete

所以我怀疑 with app.app_context(): 在删除期间被调用，试图附加和/或修改 Flask-SQLAlchemy 存储在 app< 中的 session 对象，因此整个事情都是轰炸。

Flask-SQlAlchemy 在幕后为您做了很多魔术，但您可以绕过它并直接使用 SQLAlchemy。如果您需要在删除事件期间与数据库对话，您可以创建一个新的数据库 session :

@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

但听起来您不需要这个，您只是想删除一个图像路径。在这种情况下，我会更改您的任务，使其采用一条路径:

# 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) 

希望这对您有所帮助 - 如果其中任何一项对您不起作用，请告诉我。感谢非常详细的设置，它确实有助于调试。