如何將返回列表的Celery任務鏈接到組中？

Question

我想從Celery任務返回的列表中創建一個組，以便爲​​任務結果集中的每個項目添加一個任務。

下面是一個簡單的代碼示例來解釋用例。 ???應該是上一個任務的結果。

@celery.task 
def get_list(amount): 
    # In reality, fetch a list of items from a db 
    return [i for i in range(amount)] 

@celery.task 
def process_item(item): 
    #do stuff 
    pass 

process_list = (get_list.s(10) | group(process_item.s(i) for i in ???)) 

我可能無法正確處理這個，但我敢肯定它不是安全地從任務中調用任務：

@celery.task 
def process_list(): 
    for i in get_list.delay().get(): 
     process_item.delay(i) 

我不從秒任務需要的結果。

Answer 1

您可以使用中間任務獲得此類行爲。下面是創建一個類似於您所建議的方法的「地圖」演示。

from celery import task, subtask, group 

@task 
def get_list(amount): 
    return [i for i in range(amount)] 

@task 
def process_item(item): 
    # do stuff 
    pass 

@task 
def dmap(it, callback): 
    # Map a callback over an iterator and return as a group 
    callback = subtask(callback) 
    return group(callback.clone([arg,]) for arg in it)() 

# runs process_item for each item in the return of get_list 
process_list = (get_list.s(10) | dmap.s(process_item.s())) 

信貸要問索利姆給我這個建議時，我請他幫忙在一個類似的問題。