我在哪里注册Django应用程序中的rq-scheduler作业？

Question

我想使用django_rq和rq-scheduler离线任务,但我不确定在哪里调用rq-scheduler重复任务的能力.现在,我已将我的日程安排添加到tasks.py我的应用程序中的模块,并将其导入__init__.py中.不过,必须有更好的方法来做到这一点,对吗？

提前致谢.

Answer 1

我发现运行它的最佳位置是来自你AppConfig的apps.py.

def ready(self):
    scheduler = django_rq.get_scheduler('default')

    # Delete any existing jobs in the scheduler when the app starts up
    for job in scheduler.get_jobs():
        job.delete()

    # Have 'mytask' run every 5 minutes
    scheduler.schedule(datetime.utcnow(), 'mytask', interval=60*5)

Answer 2

我创建了修改和替换的自定义管理命令rqscheduler包含在命令django_rq。此处提供了一个示例：https : //github.com/rq/rq-scheduler/issues/51#issuecomment-362352497

Answer 3

我已经__init__在我的一个项目应用程序(就Django而言)中添加了一个模块的调度,但是包含了一个小功能,可以防止两次或更多次排队.调度策略可能取决于您的特定需求(即您可能需要额外检查作业参数).

代码对我有用并符合我的需求:

import django_rq
from collections import defaultdict
import tasks

scheduler = django_rq.get_scheduler('default')

jobs = scheduler.get_jobs()
functions = defaultdict(lambda: list())

map(lambda x: functions[x.func].append(x.meta.get('interval')), jobs)

now = datetime.datetime.now()

def schedule_once(func, interval):
    """
    Schedule job once or reschedule when interval changes
    """
    if not func in functions or not interval in functions[func]\
            or len(functions[func])>1:

        # clear all scheduled jobs for this function
        map(scheduler.cancel, filter(lambda x: x.func==func, jobs))

        # schedule with new interval
        scheduler.schedule(now+datetime.timedelta(seconds=interval), func,
                interval=interval)

schedule_once(tasks.some_task_a, interval=60*5)
schedule_once(tasks.some_task_b, interval=120)

此外,我已经包装了此代码段以避免在包级别导入:

def init_scheduler():
    # paste here initialization code

init_scheduler()