Paramètres propres à la méthode run() PeriodicTask dans Celery

Question

J'écris une petite application Django et je devrais être capable de créer pour chaque objet du modèle sa tâche périodique qui sera exécutée avec un certain intervalle. Je suis utilisé pour cela une application Céleri, mais je ne peux pas comprendre une chose:

class ProcessQueryTask(PeriodicTask): 
    run_every = timedelta(minutes=1) 

    def run(self, query_task_pk, **kwargs): 
     logging.info('Process celery task for QueryTask %d' % 
query_task_pk) 
     task = QueryTask.objects.get(pk=query_task_pk) 
     task.exec_task() 
     return True 

Alors je ne suit:

>>> from tasks.tasks import ProcessQueryTask 
>>> result1 = ProcessQueryTask.delay(query_task_pk=1) 
>>> result2 = ProcessQueryTask.delay(query_task_pk=2) 

Le premier appel est le succès, mais d'autres appels périodiques retour l'erreur - TypeError: run() prend exactement 2 arguments non-mot-clé (1 donné) dans serveur celeryd. Puis-je transmettre mes propres paramètres à PeriodicTask run()?

Answer 1

Cela a été répondu à merveille par Ask Solem dans his response to your question sur le celery-users Google group.

Les tâches périodiques n'utilisent pas d'arguments, vous devez donc créer plusieurs classes ou effectuer une tâche périodique traitant plus d'un "modèle".

.: par exemple

from celery.task import PeriodicTask 
from celery.decorators import periodic_task 

# base class 
class BaseProcessQueryTask(PeriodicTask): 
    abstract = True 
    run_every = timedelta(minutes=1) 
    query_task_pk = None 

    def run(self): 
     task = QueryTask.objects.get(pk=self.query_task_pk) 
     task.exec_task() 

class ProcessQueryTask1(BaseProcessQueryTask): 
    query_task_pk = 1 

class ProcessQueryTask2(BaseProcessQueryTask): 
    query_task_pk = 2 

mais il est plus probable que voulez quelque chose comme ceci:

@task(ignore_result=True) 
def execute_query_task(task): 
    task.exec_task() 

@periodic_task(run_every=timedelta(minutes=1)) 
def process_query_tasks(): 
    for task in QueryTask.objects.all(): 
     ExecuteQueryTask.delay(task)