Каков наилучший способ реализовать context_task для period_tasks в Huey?

Question

Я ищу лучший способ реализации контекстного менеджера Peewee с периодическими задачами в Huey. У обычных задач есть такой симпатичный маленький декоратор Huey.context_task (), но похоже, что для периодических задач нет ничего похожего.

Правильно ли предположить, что мне просто придется использовать (более уродливый) оператор в периодических задачах?

Answer 1

Должно быть в состоянии сделать что-то вроде этого:

from functools import wraps

huey = RedisHuey()
db = PostgresqlDatabase(...)

def db_periodic_task(*args, **kwargs):
    def decorator(fn):
        @wraps(fn)
        def new_task():
            with db:
                fn()
        return huey.periodic_task(*args, **kwargs)(new_task)
    return decorator

@db_periodic_task(crontab('0', '0'))
def my_periodic_task():
    # db will have been opened at start.
    # db will then be closed upon return.