Comment obtenir les arguments transmis à la tâche du céleri?

Question

Pour mon application Web, j'utilise celery avec redis backend. Official FAQ vise la façon suivante pour acquérir le résultat d'une tâche si j'ai l'ID de la tâche.

result = my_task.AsyncResult(task_id) 
result.get() 

Je peux facilement accéder à des arguments dans le corps de la tâche:

@app.task 
def my_task(foo, bar, baz=None): 
    kwargs = self.request.kwargs 
    args = self.request.args 

Y at-il un moyen d'obtenir args et kwargs du AsyncResult ou où juste avoir l'identifiant de tâche?

Answer 1

Si la tâche est en attente ou si elle s'exécute actuellement, vous pouvez voir les arguments de la tâche as mentioned here.

Si vous souhaitez obtenir les arguments d'une tâche ayant échoué, vous pouvez configurer une tâche de base personnalisée avec le gestionnaire on_failure pour obtenir des arguments lorsque les tâches échouent.

from celery import Task 

class CustomBaseTask(Task):   

    def on_failure(self, exc, task_id, args, kwargs, einfo): 
     print(args) 

@app.task(base=CustomBaseTask) 
def add(x, y): 
    raise error 

D'autre part, si votre tâche est exécutée avec succès, vous aurez résultat seulement et vous ne pouvez pas récupérer vos arguments que ces informations ont été consommés et transformés par les travailleurs des courtiers.