Django Celery Workflow Chain Pausa/Ripresa

Question

C'è un modo per sospendere/riprendere un flusso di lavoro in esecuzione creato usando le catene da Celery 3.0?

Fondamentalmente, abbiamo due diversi tipi di compiti nel nostro sistema: quelli interattivi e quelli non interattivi. Per quelli non interattivi abbiamo tutti i parametri, ma quelli interattivi hanno bisogno di input da parte dell'utente. Si noti che per le attività interattive, è possibile chiedere l'input dell'utente solo dopo aver completato tutti i tak precedenti della catena, in quanto i risultati influenzeranno le attività interattive (ovvero non è possibile chiedere l'input dell'utente prima di creare la catena effettiva).

Qualche suggerimento su come affrontare questo? Davvero in perdita qui ..

idee correnti:

• Creare due sottoclassi di attività (dal sedano importazione Task). Aggiungere una variabile extra di istanza (membro della classe) alla sottoclasse di attività Interactive impostata su false per impostazione predefinita e che indica che è ancora necessario immettere parte dell'utente. In qualche modo ha accesso all'istanza del Task e lo imposta su true dall'esterno del worker di sedano (anche se l'ho guardato un po 'e non sembra possibile avere accesso agli oggetti Task direttamente da un altro modulo)
• Partiziona la catena in più catene delimitate da Lavori interattivi. Avere una sorta di meccanismo al di fuori del gestore di sedili rileva una volta che una catena ha raggiunto la sua fine e attiva il componente lato client interattivo del task interattivo. Una volta che l'utente ha inserito tutti questi dati, ottenere i dati e avviare la nuova catena in cui l'attività interattiva è alla testa della nuova catena.

Answer 1

Abbiamo implementato qualcosa come la tua seconda idea nel nostro progetto & funziona correttamente. Ecco il succo dell'implementazione.

Aggiungi nuovo campo status al tuo modello & metodo di sostituzione override.

models.py:

class My_Model(models.Model): 
    # some fields 
    status = models.IntegerField(default=0) 

    def save(self, *args, **kwargs): 
     super(My_Model, self).save(*args, **kwargs) 
     from .functions import custom_func 
     custom_func(self.status) 

tasks.py

@celery.task() 
def non_interactive_task(): 
    #do something. 

@celery.task() 
def interactive_task(): 
    #do something. 

functions.py

def custom_func(status): 
    #Change status after non interactive task is completed. 
    #Based on status, start interactive task. 

Passo status variabile per modello che è utile per visualizzare l'elemento dell'interfaccia utente affinché l'utente possa inserire le informazioni. Quando l'utente inserisce le informazioni richieste, modifica lo stato. Questo chiama custom_func che attiva il tuo interactive_task.