Łączenie wyników pracy w selerze

Question

Pracuję nad systemem aktualizacji oprogramowania przy użyciu selera. Mam przypadek użycia, który staram się wdrożyć w czysty sposób. Oto moje miejsca pracy:

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 

W czystej pytona, realizacja device_software_updates może wyglądać

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

Co jest najczystszym sposobem wdrożenia tego w Seler 3.0? Chciałbym coś zrobić, używając grup. Moja obecna implementacja wygląda następująco:

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

Niestety, oznacza to, że argspec z software_updates jest software_updates(arg_list) który nie jest idealny.

Answer 1

Wierzę, że użycie akordu byłoby właściwym sposobem na poradzenie sobie z tym.

Według dokumentacji seler na http://docs.celeryproject.org/en/latest/userguide/canvas.html#groups,

Akord to zadanie, które wykonuje tylko po wykonaniu wszystkich zadań w taskset zakończyły wykonywania.

...

Akord jest tak jak w grupie, ale z zwrotnego. Akord składa się z grupy nagłówków i ciała, gdzie ciało jest zadaniem, które powinno wykonać po zakończeniu wszystkich zadań w nagłówku.

Oto podziale, linia na przykładzie linii (z dokumentacji selera)

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

W twoim przypadku, można zrobić coś podobnego, jak:

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