Gdzie mogę zarejestrować zadanie rq-scheduler w aplikacji Django?

Question

Chciałbym używać django_rq i rq-scheduler do zadań w trybie offline, ale nie jestem pewien, gdzie zadzwonić pod numer rq-scheduler, aby zaplanować powtarzanie zadań. W tej chwili dodałem harmonogram do modułu tasks.py w mojej aplikacji i zaimportuję go w __init__ .py. Ale musi być lepszy sposób, prawda?

Z góry dziękuję.

Answer 1

Dodałem harmonogram do modułu __init__ w jednej z moich aplikacji projektowych (pod względem Django), ale zawijam z małą funkcją, która uniemożliwia kolejkowanie zadań dwa lub więcej razy. Strategia planowania może być zależna od Twoich konkretnych potrzeb (np. Możesz potrzebować dodatkowego sprawdzania argumentów zadania).

kod, który działa dla mnie i pasuje do moich potrzeb:

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) 

Również mam owinięty ten fragment, aby uniknąć importu na poziomie pakietu:

def init_scheduler(): 
    # paste here initialization code 

init_scheduler() 

Answer 2

najlepsze miejsce, w jakim znaleziono aby go uruchomić, należy to zrobić pod adresem AppConfig pod numerem .

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 3

stworzyłem polecenia zarządzania niestandardową, która zmienia i zastępuje komenda rqscheduler zawarte w django_rq. Przykład znajduje się tutaj: https://github.com/rq/rq-scheduler/issues/51#issuecomment-362352497