Combinazione dei risultati del lavoro in sedano

Question

Sto lavorando a un sistema di aggiornamento software utilizzando il sedano. Ho un caso d'uso che sto faticando a implementare in modo pulito. Qui sono i miei lavori:

device_software_updates (device_id)

returns a list of software updates that need to be installed on a device 

installed_device_software (device_id)

returns the software modules that are currently installed on a device 

latest_device_software (device_id)

returns the latest software versions available for a device 

software_updates (installed_software, latest_software)

returns the latest software modules that are not installed 

in puro Python, l'attuazione di device_software_updates potrebbe assomigliare

def device_software_updates(device_id): 
    return software_updates(installed_device_software(device_id), 
          latest_device_software(device_id)) 

Qual è il modo più pulito per implementare questo in Sedano 3.0? Mi piacerebbe fare qualcosa usando i gruppi. Il mio attuale implementazione si presenta così:

def device_software_updates(device_id): 
    return (
     group(installed_device_software.s(device_id), 
       latest_device_software.s(device_id)) | 
     software_updates.s() 
    )() 

Sfortunatamente, questo significa che l'argspec di software_updates è software_updates(arg_list) che non è ideale.

Answer 1

Credo che l'uso di una corda sarebbe il modo giusto per gestire questa situazione.

Secondo la documentazione del Sedano a http://docs.celeryproject.org/en/latest/userguide/canvas.html#groups,

Un accordo è un compito che viene eseguito solo dopo che tutte le attività in un taskset hanno terminato l'esecuzione.

...

Un accordo è proprio come un gruppo, ma con un callback. Un accordo consiste in un gruppo di intestazioni e un corpo, in cui il corpo è un'attività che deve essere eseguita dopo che tutte le attività nell'intestazione sono state completate.

Ecco una panne, riga per esempio la linea (dalla documentazione di sedano)

callback = tsum.subtask() 
header = [add.subtask((i, i)) for i in xrange(100)] 
result = chord(header)(callback) 
result.get() 

Nel tuo caso, si potrebbe fare qualcosa di simile, come:

@celery.task 
def device_software_updates(): 
    callback = software_updates.subtask() 
    header = [ 
       installed_device_software.subtask(device_id), 
       latest_device_software.s(device_id) 
      ] 
    result = chord(header)(callback) 
    return result.get()