在芹菜任务中反映SQLAlchemy元数据？
出于更好的可测试性和其他原因，最好将SQLAlchemy数据库会话配置为非全局的，如以下问题中所述：
how to setup sqlalchemy session in celery tasks with no global variable (在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)。
我正在考虑使用alembic版本更改作为重新反映的信号，但不太确定如何让它在芹菜中很好地工作。例如，如果多个工作线程将立即感知到更改，则可以以非线程安全的方式处理全局meta。
如果这很重要，芹菜在案例中的使用是独立的，没有web框架模块/应用程序/任何存在于芹菜应用程序中的东西。问题也被简化了，因为只有SQLAlchemy核心在使用，而不是对象映射器。
回答 1
Stack Overflow用户
发布于 2019-05-22 03:11:46
这只是部分解决方案，它是针对核心对象关系管理的(但我猜对于SQLAlchemy来说，类似的东西很容易实现)。
要点：
- engine位于模块级，但config (访问URL，参数)来自于它自己的工厂函数
- in module level：
BaseModel = automap_base()，然后表类使用该BaseModel作为超类，通常只有一个参数-__tablename__，但可以在其中添加任意关系、属性(非常类似于正常使用的
)模块级的
- ：
__tablename__
测试(使用pytest)在模块级注入conftest.py中的环境变量(例如DB_URL)。
一个重要时刻：database_session总是在任务函数中初始化(即调用工厂函数)，并显式地传播到所有函数中。这种方式允许自然地控制工作单元，通常每个任务一个事务。这也简化了测试，因为所有使用数据库的函数都可以提供假的或真实的(测试)数据库会话。
“任务函数”是上面的一个函数，它是在函数中调用的，它是由任务来修饰的-这样任务函数就可以在没有任务机器的情况下进行测试。
这只是部分解决方案，因为重做反射是不存在的。如果任务工作者可以停止片刻(数据库无论如何都会因为模式更改而停机)，因为这些通常是后台任务，所以不会造成问题。Worker还可以由一些外部watchdog重新启动，它可以监视数据库更改。这可以通过使用supervisord或一些其他方法来方便地控制在前台运行的芹菜工人。
总而言之，在我如上所述解决了这个问题之后，我更加看重“显性优于隐性”的哲学。所有这些神奇的“应用”，无论是芹菜还是Flask中的请求，可能会在函数签名中带来极小的缩写，但我宁愿将某种上下文传递到调用链中，以提高可测试性，并更好地理解和管理上下文。
https://stackoverflow.com/questions/52570671
复制相似问题
腾讯云开发者
