Comment gérer les tâches dans Céleri?

Question

J'ai une tâche avec un certain paramètre, et je veux savoir s'il y a une tâche avec le même paramètre. Je donne les résultats suivants:

@task 
def some_task(id): 
    some_task.update_state(state="PROGRESS", meta={"id": id}) 
    some_action_by_id(id) 

Mais je voudrais ceci:

@task 
def some_task(id): 
    if !check_task(id): 
    some_task.update_state(state="PROGRESS", meta={"id": id}) 
    some_action_by_id(id) 

Comment puis-je y arriver?

Answer 1

Vous voulez probablement dire que vous ne voulez qu'une "certaine_tâche" s'exécutant pour chaque identifiant unique à chaque instant. Vous devez donc mettre en place un mécanisme de verrouillage. Jetez un oeil here. Céleri joue super avec Redis!

Answer 2

voici ma solution:

from celery.task.control import inspect 
from celery.result import AsyncResult 

def get_same_task(aTaskName, aArgs, aHosts): 
    for jobs in aHosts.values(): 
     for job in jobs: 
      if job['name'] == aTaskName and job['args'] == str(aArgs): 
       return job['id'] 
    return None 

class IgnoreSameArgumentsTask(Task): 
    abstract = True 
    inspect = inspect() 

    def delay(self, *args, **kwargs): 
     vHosts_Jobs = self.inspect.active() 
     vTaskId = get_same_task(self.name, args, vHosts_Jobs) 
     if vTaskId != None: 
      return AsyncResult(vTaskId) 
     else: 
      return super(IgnoreSameArgumentsTask, self).delay(*args, **kwargs) 

@celery.task(base=IgnoreSameArgumentsTask) 
def add(x, y): 
    sleep(x+y) 
    return x + y