Huey周期任务内的SQLAlchemy会话

Question

我正在将Huey与一个简单的金字塔应用程序集成。我没有在应用程序中使用全局SQLAlchemy会话(我正在使用最新的alchemy scaffold)。然而，似乎没有其他直接的方法来为周期性任务提供会话。

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是否提供钩子在任务完成时关闭DB连接?为这些任务提供线程安全连接的最佳方式是什么?

提前感谢!

Answer 1

您可以在tasks中使用工厂构建会话对象:

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

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

Answer 2

scoped_session为您提供了一个上下文/线程本地会话(即，它对应于每个线程中的单个DB连接，并且当您需要每个非线程的单独会话时，也可以配置自定义作用域。

所以，基本上，你所需要做的就是有一个正确配置的伪全局变量(类似于你现在拥有的)，并确保在任务开始时调用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(...)