将 SQLAlchemy 会话与 Celery 一起使用的正确方法是什么？

Question

我一直试图弄清楚这一点，但我现在能找到的很多答案都是过时的（来自 6 年前的帖子），或者不太相关。

问题实际上是如何在 celery 中正确处理数据库会话。我当前的设置是：我有一个包含dsn、engine 和Session 的全局DbEngine 对象。所以每次如果我想使用会话，我只需调用sess = db.Session() 并开始在我的 Flask 应用程序中使用它们。如下所示：

#db.py
class DbEngine:
  def __init__(self, path, ...):
    self.dsn = self.create_dsn_from_file(path)
    self.engine = create_engine(self.dsn)
    self.Session = scoped_session(sessionmaker(bind=self.engine))

在我将 celery 引入我的应用程序之前，我经常遇到各种错误（协议错误、sslSocket 错误等）。而且我无法在本地重现它们，如果我只是为这些任务添加重试，它通常会得到修复（它通常会通过 3 次重试）。所以我怀疑这可能是由共享会话引起的。

然后我决定为 celery 应用更改会话：每次如果我需要会话，我实际上会创建一个新引擎，创建一个新 Session，并返回新创建的Session()。然而，类似但不完全相同的问题再次发生（各种协议错误，代码不同）。

我看到 celery 有自己的 SessionManager，但是我找不到使用示例代码。我在考虑以下结构：

# celery_app.py

celery = ... 

session_manager = SessionManager()
engine, Session = session_manager.create_session(dsn)

在我所做的任务中：

# task_1.py
from celery_app import celery, Session

@celery.task
def tsk():
  sess = Session()
  sess.query(...)
  ... 
  sess.close()

但真的不确定这是否是预期的方法，因为我只调用了一次 .create_session()，而且我不知道何时以及如何调用 SessionManager 中的其他函数。

作为背景，我使用的是 DB2。

如果有人知道如何正确使用 SessionManager 或有使用 SQLAlchemy 的会话经验，如果您能提供任何见解，将不胜感激。提前谢谢！

Answer 1

也许不是您正在寻找的答案，但flask_sqlalchemy 在 SQLAlchemy 上提供了一个对 Flask 友好的层，几乎完全消除了手动管理会话的需要。一旦您按照Flask docs 中的示例设置了 ContextTask，它就可以很好地与 celery 搭配使用。