Kết hợp kết quả công việc trong cần tây

Question

Tôi đang làm việc trên một hệ thống nâng cấp phần mềm sử dụng cần tây. Tôi có một trường hợp sử dụng mà tôi đang đấu tranh để thực hiện sạch sẽ. Dưới đây là công việc của tôi:

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 

Trong trăn tinh khiết, việc thực hiện device_software_updates có thể trông như

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

cách sạch để thực hiện điều này trong cần tây là gì 3.0? Tôi muốn làm một cái gì đó bằng cách sử dụng các nhóm. thực hiện hiện tại của tôi trông như thế này:

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

Thật không may, điều này có nghĩa rằng argspec của software_updates là software_updates(arg_list) mà không phải là lý tưởng.

Answer 1

Tôi tin rằng việc sử dụng hợp âm là sẽ là cách phù hợp để xử lý việc này.

Theo các tài liệu của Cần tây tại http://docs.celeryproject.org/en/latest/userguide/canvas.html#groups,

Một âm là một nhiệm vụ mà chỉ thực hiện sau khi tất cả các nhiệm vụ trong một taskset đã hoàn tất thi công.

...

Hợp âm giống như một nhóm nhưng có gọi lại. Hợp âm bao gồm nhóm tiêu đề và nội dung, trong đó nội dung là nhiệm vụ cần thi hành sau khi tất cả các tác vụ trong tiêu đề hoàn tất.

Đây là một chia nhỏ, dòng bằng ví dụ dòng (từ các tài liệu cần tây)

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

Trong trường hợp của bạn, bạn có thể làm điều gì đó tương tự, như:

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