在芹菜任务中反映出SQLAlchemy元数据？

Question

为了更好的可测试性和其他原因,最好在下面的问题中非常好地描述SQLAlchemy数据库会话配置非全局:

如何在没有全局变量的芹菜任务中设置sqlalchemy会话(并在https://github.com/celery/celery/issues/3561中讨论)

现在,问题是,如何优雅地处理元数据？如果我的理解是正确的,元数据可以有一次,例如:

engine = create_engine(DB_URL, encoding='utf-8', pool_recycle=3600,
                       pool_size=10)
# db_session = get_session()  # this is old global session
meta = MetaData()
meta.reflect(bind=engine)

反映每个任务执行不利于性能原因,元数据或多或少是稳定的和线程安全的结构(如果我们只读它).

但是,元数据有时会发生变化(celery不是db模式的"所有者"),从而导致工作者出错.

什么可以meta以一种可测试的方式处理的优雅方式,还能够对底层数据库更改做出反应？(使用中的蒸馏器,如果相关的话).

我正在考虑使用alembic版本更改作为重新反映的信号,但不太确定如何使其在芹菜中很好地工作.例如,如果不止一个工作人员将立刻感知到更改,则meta可以以非线程安全方式处理全局.

如果重要的话,芹菜使用的情况是独立的,没有网络框架模块/应用程序/芹菜应用程序中的任何内容.问题也得到了简化,因为只使用了SQLAlchemy Core,而不是对象映射器.

Answer 1

这只是部分解决方案，它适用于 SQLAlchemy ORM（但我想类似的东西对于 Core 来说很容易实现）。

要点：

• 引擎位于模块级别，但配置（访问 URL、参数）来自os.environ
• 会话位于它自己的工厂函数中
• 在模块级别：BaseModel = automap_base()然后表类使用 BaseModel 作为超类，通常只有一个参数 - __tablename__，但可以在那里添加任意关系、属性（与正常的 ORM 使用非常相似）
• 在模块级别：BaseModel.prepare(ENGINE, reflect=True)

测试（使用 pytest）在模块级别注入环境变量（例如DB_URL） 。conftest.py

一个重要的时刻：database_session始终在任务函数中启动（即调用工厂函数），并显式传播到所有函数中。这种方式允许自然地控制工作单元，通常每个任务一个事务。这也简化了测试，因为所有使用数据库的功能都可以提供假的或真实的（测试）数据库会话。

“任务函数”就是上面是一个函数，在函数中调用，用task来装饰——这样就可以在没有任务机器的情况下测试任务函数。

这只是部分解决方案，因为不存在重做反射。如果任务工作人员可以停止一会儿（并且数据库无论如何都会由于架构更改而经历停机），因为这些通常是后台任务，所以它不会造成问题。工作人员还可以通过一些外部看门狗重新启动，该看门狗可以监视数据库更改。通过使用 Supervisord 或其他方式来控制在前台运行的 celery 工作人员可以方便地实现这一点。

总而言之，在我解决了上述问题之后，我更加看重“显性优于隐性”的哲学。所有这些神奇的“应用程序”，“请求”，无论是在 celery 还是 Flask 中，都可能在函数签名中带来微小的缩写，但我宁愿在调用链中传递某种上下文，以提高可测试性和更好的上下文理解，管理。