1

我正在将Huey与一个简单的金字塔应用程序集成。我没有在应用程序中使用全局 SQLAlchemy 会话（我正在使用最新的炼金术脚手架）。但是，似乎没有其他直接的方法可以为周期性任务提供会话。

from huey import RedisHuey

huey = RedisHuey(password=os.environ.get('REDIS_PASSWORD', ''))
DBSession = scoped_session(sessionmaker())


@huey.periodic_task(crontab(minute='*/1'))
def notify_not_confirmed_assignments():
    # TODO: Use a non-global DB session
    assignments = DBSession.query(Assignment).filter_by(date=next_date).all()

Huey 是否提供挂钩以在任务完成时关闭数据库连接？为这些任务提供线程安全连接的最佳方式是什么？

提前致谢！

4

2 回答 2

1

您可以在任务中使用工厂构建会话对象：

factory = sessionmaker()
factory.configure(bind=engine)
session = factory()

无需使用作用域会话，只需初始化引擎并将其传递给工厂。

于 2016-10-26T18:42:39.607 回答
0

scoped_session为您提供上下文/线程本地会话（即，它对应于每个线程中的单个数据库连接，并且当您需要每个不是线程的单独会话时，也可以配置自定义范围。

因此，基本上，您需要做的就是拥有一个正确配置的伪全局变量（类似于您现在拥有的）并确保在任务开始时调用 DBSession.begin() 和 DBSession.commit()最后 - 手动执行可能是一件苦差事，但它可以很容易地抽象为上下文管理器

def my_task():
    with magically_start_session() as session:
        session.query(...)

或进入装饰器：

@huey.periodic_task(crontab(minute='*/1'))
@start_session
def notify_not_confirmed_assignments(session):
    # TODO: Use a non-global DB session
    assignments = session.query(...)
于 2016-10-26T21:44:59.347 回答