Combinación de resultados de trabajo en apio

Question

Estoy trabajando en un sistema de actualización de software con apio. Tengo un caso de uso que estoy luchando por implementar de forma limpia. Aquí están mis trabajos:

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 

en Python puro, la implementación de device_software_updates puede tener un aspecto como

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

¿Cuál es la forma más limpia de implementar esto en Apio 3.0? Me gustaría hacer algo usando grupos. Mi implementación actual se parece a esto:

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

Desafortunadamente, esto significa que el argspec de software_updates es software_updates(arg_list) que no es ideal.

Answer 1

Creo que usar un acorde sería la forma correcta de manejar esto.

Según la documentación del apio en http://docs.celeryproject.org/en/latest/userguide/canvas.html#groups,

Un acorde es una tarea que sólo se ejecuta después de que todas las tareas de una taskset han terminado de ejecutarse.

...

Una cuerda es igual a un grupo pero con una devolución de llamada. Un acorde consiste en un grupo de encabezado y un cuerpo, donde el cuerpo es una tarea que se debe ejecutar después de que se completen todas las tareas del encabezado.

Aquí es un desglose, línea por ejemplo de línea (a partir de la documentación de apio)

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

En su caso, se podría hacer algo similar, como:

@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()