从 Crontab 开始获取简单 Huey 示例的挑战

Challenges getting simple Huey example started with Crontab

我正在尝试实施 suggested organization of code from the Huey docs into an existing app, and also following the simple example。目标是构建一个 crontab,每天 运行 上午 3:00 执行任务。

我启动了两个终端选项卡,第一个带有消费者 运行示例中的脚本:

PYTHONPATH=.:$PYTHONPATH
export WORKER_CLASS=${1:-thread}
huey_consumer.py main.huey --workers=4 -k $WORKER_CLASS -C -S

然后,在另一个选项卡中,我 运行 main.py 脚本:

python main.py

config.py

from huey import SqliteHuey

huey = SqliteHuey(filename='/tmp/huey.db')

tasks.py

from config import huey

# Note that this time is 1 minute before whenever I'm debugging.
# I'm using the 3 am example as what we're aiming for in our final product.
@huey.periodic_task(crontab(minute='0', hour='3'))
def run_this_function():
    system = New_Class() # Class instantiation since there's a bunch that happens when a new class is set up 
    system.run_method # Runs a bunch of methods from the class in one location

main.py

from config import huey
from tasks import run_this_function

def main:
    run_this_function()

if __name__ == "__main__":
    main()

任务 运行 立即完成,因为我是 Huey 的新手,不确定我可能缺少什么才能使其按计划进行。我尝试了很多 crontab 组合,不确定是否存在挑战,或者我如何在 main 方法中调用 run_this_function。感谢您的帮助!

所以首先,您实际上并不想自己调用 run_this_function(),因此没有必要 运行 您的 main.py 脚本在另一个选项卡中。您只需要让 huey 消费者 运行ning 因为您希望它负责在请求的时间执行任务。

您的消费者需要能够发现该任务,您可以通过在执行时将其导入该主文件然后通过该文件启动 huey 实例(您也正在执行)来实现这一点。一个简单的测试可能是将打印语句放在与定义周期性任务的位置相同的文件中。一旦你 运行 消费者,你应该看到你的打印声明。否则,您的任务也不会被消费者接收。

其次,我建议不要使用crontab功能。我过去一直没能让它工作,相反,你最好编写自己的函数来配置凌晨 3 点。 Huey 使用当前时间戳定期调用提供的函数。因此,作为一个可重用的示例,您可以执行以下操作:

def run_at_my_preferred_hour(hour_to_run):
    last_day_it_ran_for = None

    def check_if_task_needs_to_run_for_this_time(current_timestamp):
        if current_timestamp.hour == hour_to_run and current_timestamp.day != last_day_it_ran_for:
            # Save that you already ran the task for this day
            last_day_it_ran_for = current_timestamp.day
            return True
        return False

    return check_if_task_needs_to_run_for_this_time

    
@huey.periodic_task(run_at_my_preferred_hour(3))
def run_this_function():
   ...