Sedano - gruppo all'interno di una catena di

Question

voglio usare un gruppo (o blocchi) all'interno di una catena, come:

chain(getRange.s(3), GROUP() , xsum.s()) 

Dove GROUP() è un gruppo di double() compiti, vale a dire group(double(0),double(1),double(2)). Una domanda simile è stata pubblicata in How to chain a Celery task that returns a list into a group? ma non viene spiegato come passare l'output dal gruppo all'attività successiva nella catena.

@task 
def getRange(x): 
    return range(x) 

@task 
def double(nr): 
    return nr*2 

@task 
def xsum(list): 
    return sum(list) 

Answer 1

Non credo che ci sia un modo per farlo con gli attuali primitivi in ​​una singola catena. Passare i callback come nella domanda che hai menzionato non ti consentirà di ascoltare quando le attività di gruppo avranno terminato. Il più vicino si può ottenere è qualcosa di simile:

@task 
def get_range(x): 
    return range(x) 

@task 
def mapper(nr): 
    return nr * 2 

@task 
def reducer(nrs): 
    return sum(nrs) 

@task 
def double_then_sum(nrs): 
    return (
    group([mapper.s(nr) for nr in nrs]) | 
    reducer.s() 
)() 

ar = (get_range.s(3) | double_then_sum.s())() # call the procedure 
ar.result.result # get the result 

altrimenti si potrebbe provare a utilizzare dynamic chaining, che porterebbe ad una soluzione più semplice, o semplicemente usare map se non è necessario le attività raggruppate per l'esecuzione in parallelo.