एक सेलेरी कार्य को कैसे श्रृंखलाबद्ध करें जो समूह में एक सूची देता है?

Question

मैं एक सेलेरी कार्य द्वारा लौटाई गई सूची से समूह बनाना चाहता हूं, ताकि कार्य में प्रत्येक आइटम के लिए सेट किया गया हो, समूह में एक कार्य जोड़ा जाएगा।

उपयोग केस की व्याख्या करने के लिए यहां एक सरल कोड उदाहरण है। ??? पिछले कार्य से परिणाम होना चाहिए।

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

मुझे इस सुझाव देने के लिए सोमाल से पूछने के लिए क्रेडिट जब मैंने उसे इसी तरह की समस्या पर मदद के लिए कहा।