Jak powiązać zadanie z selerem, które zwraca listę do grupy?

Question

Chcę utworzyć grupę z listy zwróconej przez zadanie Celery, aby dla każdego elementu w zestawie wyników zadania jedno zadanie zostało dodane do grupy.

Oto prosty przykład kodu wyjaśniający przypadek użycia. Numer ??? powinien być wynikiem poprzedniego zadania.

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

Jestem prawdopodobnie nie zbliża to poprawnie, ale jestem pewien, że nie jest to bezpieczne, aby zadzwonić do zadań z poziomu zadania:

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

nie muszę wynik z zadania sekund .

Answer 1

Możesz uzyskać tego rodzaju zachowanie za pomocą zadania pośredniego. Oto demonstracja tworzenia "podobnej do mapy" metody, która działa tak, jak zasugerowałeś.

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

Chciałbym prosić Solema o udzielenie mi tej sugestii, gdy poprosiłem go o pomoc w podobnej sprawie.