How to chain a Celery task that returns a list into a group?

by

You can get this kind of behavior using an intermediate task. Here’s a demonstration of creating a “map” like method that works like you’ve suggested.

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

Credit to Ask Solem for giving me this suggestion when I asked him for help on a similar issue.

More Related Contents:

  1. Retrieve list of tasks in a queue in Celery
  2. How to dynamically add / remove periodic tasks to Celery (celerybeat)
  3. Cancel an already executing task with Celery?
  4. How to run celery on windows?
  5. How to check task status in Celery?
  6. Running “unique” tasks with celery
  7. Reporting yielded results of long-running Celery task
  8. Celery Received unregistered task of type (run example)
  9. How can I schedule a Task to execute at a specific time using celery?
  10. Distributed task queues (Ex. Celery) vs crontab scripts
  11. Understanding celery task prefetching
  12. Running Scrapy spiders in a Celery task
  13. Celery: is there a way to write custom JSON Encoder/Decoder?
  14. Django Celery Logging Best Practice
  15. Detect whether Celery is Available/Running
  16. How to combine Celery with asyncio?
  17. What’s the pythonic way to use getters and setters?
  18. How to start a background process in Python?
  19. Generate random numbers summing to a predefined value
  20. ‘str’ object does not support item assignment
  21. Iterate over object attributes in python [duplicate]
  22. How to print one character at a time on one line?
  23. Expand the line with specified width in data unit
  24. Why does next raise a ‘StopIteration’, but ‘for’ do a normal return?
  25. Checking email with Python
  26. Why is Collections.counter so slow?
  27. Passing a column name in a SELECT statement in Python
  28. How to convert hexadecimal string to bytes in Python?
  29. Multi Index Sorting in Pandas
  30. matplotlib taking time when being imported

Leave a Comment