用芹菜进行SQLAlchemy治疗的正确方法是什么？

Question

我已经试着弄明白这一点有一段时间了，但我现在能找到的很多答案都过时了（6年前的帖子），或者说相关程度更低

问题是如何在芹菜中正确处理数据库会话。我当前的设置是：我有一个全局DbEngine对象，它包含dsn、engine和Session。因此，每次如果我想使用会话，我只需调用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))

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

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

我看到芹菜有自己的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 docs中的示例中设置ContextTask后，它可以很好地处理芹菜