为了更好的可测试性和其他原因,最好让SQLAlchemy数据库会话配置非全局,如以下问题中所述:
如何在没有全局变量的芹菜任务中设置 SQLlChemy 会话(也在 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)
出于性能原因,反映每个任务执行并不好,元数据或多或少是稳定和线程安全的结构(如果我们只阅读它(。
但是,元数据有时会更改(芹菜不是数据库架构的"所有者"(,从而导致工作线程错误。
以可测试的方式处理meta
,并且仍然能够对底层数据库更改做出反应,这是什么优雅的方法?(使用中的alembic,如果相关的话(。
我正在考虑使用alembic版本更改作为重新反映的信号,但不太确定如何使其在芹菜中很好地工作。例如,如果多个工作线程同时感知到更改,则可以以非线程安全方式处理全局meta
。
如果重要的话,在这种情况下,芹菜的使用是独立的,没有 Web 框架模块/应用程序/芹菜应用程序中存在的任何内容。这个问题也得到了简化,因为只有SQLAlchemy Core正在使用中,而不是对象映射器。
这只是部分解决方案,它适用于SQLAlchemy ORM(但我想类似的东西很容易实现为Core(。
大意:
- 引擎在模块级别,但配置(访问 URL、参数(来自
os.environ
- 会话在其自己的工厂函数中
- 在模块级别:
BaseModel = automap_base()
然后表类使用该 BaseModel 作为超类,通常只有一个参数 -__tablename__
,但可以在那里添加任意关系、属性(与正常的 ORM 使用非常相似( - 在模块级别:
BaseModel.prepare(ENGINE, reflect=True)
测试(使用 pytest(在模块级别的conftest.py
中注入环境变量(例如DB_URL
(。
一个重要时刻:database_session
始终在任务函数中启动(即调用工厂函数(,并显式传播到所有函数中。这种方式允许自然地控制工作单元,通常每个任务一个事务。这也简化了测试,因为所有使用数据库的函数都可以通过假的或真实的(测试(数据库会话提供。
"任务函数"就是上面就是一个函数,在函数中调用,由任务装饰——这样任务函数就可以在没有任务机器的情况下进行测试。
这只是部分解决方案,因为不存在重做反射。如果任务工作线程可以停止片刻(并且数据库无论如何都会因架构更改而经历停机(,因为这些通常是后台任务,因此不会造成问题。工作线程也可以由一些外部监视器重新启动,该监视器可以监视数据库更改。这可以通过使用主管或其他方法来控制在前台运行的芹菜工人来方便。
总而言之,在我解决了上述问题之后,我更加重视"显式优于隐性"的哲学。所有这些神奇的"应用程序","请求"无论是芹菜还是 Flask,都可能会在函数签名中带来微小的缩写,但我宁愿将某种上下文传递到调用链中以提高可测试性并更好地理解和管理上下文。