Làm thế nào để chuỗi một nhiệm vụ Celery trả về một danh sách thành một nhóm?

Question

Tôi muốn tạo một nhóm từ danh sách được trả về bởi tác vụ Celery, để cho mỗi mục trong tập kết quả nhiệm vụ, một nhiệm vụ sẽ được thêm vào nhóm.

Dưới đây là một ví dụ mã đơn giản để giải thích trường hợp sử dụng. ??? phải là kết quả từ nhiệm vụ trước đó.

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

Tôi có lẽ không tiếp cận này một cách chính xác, nhưng tôi khá chắc chắn nó không phải là an toàn để gọi nhiệm vụ từ bên trong nhiệm vụ:

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

Tôi không cần kết quả từ nhiệm vụ giây .

Answer 1

Bạn có thể nhận loại hành vi này bằng cách sử dụng tác vụ trung gian. Dưới đây là một minh chứng về việc tạo một "bản đồ" giống như phương pháp hoạt động như bạn đã đề xuất.

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

Tín dụng để hỏi Cung cấp cho tôi đề nghị này khi tôi yêu cầu trợ giúp về vấn đề tương tự.