Verificare se esiste un'attività di sedano

Question

È possibile scoprire se esiste un'attività con un determinato ID attività? Quando cerco di ottenere lo stato, sarò sempre in attesa.

>>> AsyncResult('...').status 
'PENDING' 

Desidero sapere se un determinato ID di attività è un vero ID di compito celery e non una stringa casuale. Voglio risultati diversi a seconda che esista un'attività valida per un determinato ID.

Potrebbe essersi verificato un compito valido in passato con lo stesso ID ma i risultati potrebbero essere stati eliminati dal back-end.

Answer 1

sedano non scrive uno stato in cui l'attività viene inviato, ciò è in parte un'ottimizzazione (vedi http://docs.celeryproject.org/en/latest/userguide/tasks.html#state).

Se si ha realmente bisogno, è semplice aggiungere:

from celery import current_app 
# `after_task_publish` is available in celery 3.1+ 
# for older versions use the deprecated `task_sent` signal 
from celery.signals import after_task_publish 

@after_task_publish.connect 
def update_sent_state(sender=None, body=None, **kwargs): 
    # the task may not exist if sent using `send_task` which 
    # sends tasks by name, so fall back to the default result backend 
    # if that is the case. 
    task = current_app.tasks.get(sender) 
    backend = task.backend if task else current_app.backend 

    backend.store_result(body['id'], None, "SENT") 

Quindi è possibile verificare lo stato di ATTESA per rilevare che un compito non è (apparentemente) stato inviato:

>>> result.state != "PENDING" 

Answer 2

Prova

AsyncResult('blubb').state 

che può funzionare.

Dovrebbe restituire qualcosa di diverso.

Answer 3

È necessario chiamare .get() sull'oggetto AsyncTask creato per recuperare effettivamente il risultato dal back-end.

Vedere Celery FAQ.

---

Per chiarire ulteriormente la mia risposta.

Qualsiasi stringa è tecnicamente un ID valido, non esiste un modo per convalidare l'ID dell'attività. L'unico modo per scoprire se esiste un'attività consiste nel chiedere al back-end se ne è a conoscenza e per farlo è necessario utilizzare .get().

Questo introduce il problema che i blocchi .get() quando il back-end non ha alcuna informazione sull'ID di attività che hai fornito, questo è di progettazione per consentire di avviare un'attività e quindi attendere il suo completamento.

Nel caso della domanda originale, assumerò che l'OP desideri ottenere lo stato di un'attività precedentemente completata. Per fare questo si può passare un piccolissimo timeout e timeout cattura errori:

from celery.exceptions import TimeoutError 
try: 
    # fetch the result from the backend 
    # your backend must be fast enough to return 
    # results within 100ms (0.1 seconds) 
    result = AsyncResult('blubb').get(timeout=0.1) 
except TimeoutError: 
    result = None 

if result: 
    print "Result exists; state=%s" % (result.state,) 
else: 
    print "Result does not exist" 

Va da sé che questa unica opera se il back-end è la memorizzazione dei risultati, se non è non c'è modo di sapere se un ID attività è valido o no perché nulla tiene traccia di loro.

---

Ancora più chiarificazione.

Non è possibile eseguire ciò che si desidera fare utilizzando il backend AMQP perché it does not store results, it forwards them.

Il mio suggerimento sarebbe di passare a un back-end del database in modo che i risultati siano in un database che è possibile interrogare al di fuori dei moduli di sedano esistenti. Se nessuna attività esiste nel database dei risultati, è possibile assumere che l'ID non è valido.

Answer 4

Per favore correggimi se sbaglio.

if built_in_status_check(task_id) == 'pending' 
    if registry_exists(task_id) == true 
     print 'Pending' 
    else 
     print 'Task does not exist' 

Answer 5

AsyncResult.state restituisce PENDING in caso di ID attività sconosciute.

ATTESA

un'attività è in attesa per l'esecuzione o sconosciuto. Qualsiasi ID attività che non sia noto è implicito nello stato in sospeso.

http://docs.celeryproject.org/en/latest/userguide/tasks.html#pending

È possibile fornire ID di attività personalizzati, se avete bisogno di distinguere gli ID sconosciuti da quelli esistenti:

>>> from tasks import add 
>>> from celery.utils import uuid 
>>> r = add.apply_async(args=[1, 2], task_id="celery-task-id-"+uuid()) 
>>> id = r.task_id 
>>> id 
'celery-task-id-b774c3f9-5280-4ebe-a770-14a6977090cd' 
>>> if not "blubb".startswith("celery-task-id-"): print "Unknown task id" 
... 
Unknown task id 
>>> if not id.startswith("celery-task-id-"): print "Unknown task id" 
... 

Answer 6

Proprio ora sto usando seguente schema:

1. Ottieni un ID attività.
2. Impostato su chiave memcache come 'task_% s'% task.id messaggio 'Avviato'.
3. Passare ID attività al client.
4. Ora dal client posso monitorare lo stato dell'attività (impostato dai messaggi di attività a memcache).
5. Dall'attività alla consegna - impostare il messaggio chiave memcache "Pronto".
6. Da client su attività pronta: avvia un'attività speciale che cancellerà la chiave da memcache ed eseguirà le azioni di pulizia necessarie.