结合芹菜的工作成果

Question

我正在使用芹菜的软件升级系统.我有一个用例,我正在努力实现干净利落.这是我的工作:

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

在纯python中,device_software_updates的实现可能看起来像

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

在Celery 3.0中实现这一目标的最简洁方法是什么？我想用小组做点什么.我目前的实现如下:

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

不幸的是,这意味着software_updates的argspec software_updates(arg_list)并不理想.

Answer 1

我相信使用和弦将是处理这个问题的正确方法.

根据http://docs.celeryproject.org/en/latest/userguide/canvas.html#groups上的Celery文档,

和弦是仅在任务集中的所有任务完成执行后执行的任务.

...

和弦就像一个群体,但有一个回调.和弦由标题组和正文组成,其中正文是在标题中的所有任务完成后应执行的任务.

这是一个细分的逐行示例(来自Celery文档)

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

在你的情况下,你可以做类似的事情,如:

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