How to combine Celery with asyncio?

by

EDIT: 01/12/2021 previous answer (find it at the bottom) didn’t age well therefore I added a combination of possible solutions that may satisfy those who still look on how to co-use asyncio and Celery

Lets quickly break up the use cases first (more in-depth analysis here: asyncio and coroutines vs task queues):

  • If the task is I/O bound then it tends to be better to use coroutines and asyncio.
  • If the task is CPU bound then it tends to be better to use Celery or other similar task management systems.

So it makes sense in the context of Python’s “Do one thing and do it well” to not try and mix asyncio and celery together.

BUT what happens in cases where we want to be able to run a method both asynchronously and as an async task? then we have some options to consider:

  • The best example that I was able to find is the following: https://johnfraney.ca/posts/2018/12/20/writing-unit-tests-celery-tasks-async-functions/ (and I just found out that it is @Franey’s response):

    1. Define your async method.

    2. Use asgiref‘s sync.async_to_sync module to wrap the async method and run it synchronously inside a celery task:

      # tasks.py
import asyncio
from asgiref.sync import async_to_sync
from celery import Celery

app = Celery('async_test', broker="a_broker_url_goes_here")

async def return_hello():
    await asyncio.sleep(1)
    return 'hello'


@app.task(name="sync_task")
def sync_task():
    async_to_sync(return_hello)()

  • A use case that I came upon in a FastAPI application was the reverse of the previous example:

    1. An intense CPU bound process is hogging up the async endpoints.

    2. The solution is to refactor the async CPU bound process into a celery task and pass a task instance for execution from the Celery queue.

    3. A minimal example for visualization of that case:

      import asyncio
import uvicorn

from celery import Celery
from fastapi import FastAPI

app = FastAPI(title="Example")
worker = Celery('worker', broker="a_broker_url_goes_here")

@worker.task(name="cpu_boun")
def cpu_bound_task():
    # Does stuff but let's simplify it
    print([n for n in range(1000)])

@app.get('/calculate')
async def calculate():
    cpu_bound_task.delay()

if __name__ == "__main__":
    uvicorn.run('main:app', host="0.0.0.0", port=8000)

  • Another solution seems to be what @juanra and @danius are proposing in their answers, but we have to keep in mind that performance tends to take a hit when we intermix sync and async executions, thus those answers need monitoring before we can decide to use them in a prod environment.

Finally, there are some ready-made solutions, that I cannot recommend (because I have not used them myself) but I will list them here:

  • Celery Pool AsyncIO which seems to solve exactly what Celery 5.0 didn’t, but keep in mind that it seems a bit experimental (version 0.2.0 today 01/12/2021)
  • aiotasks claims to be “a Celery like task manager that distributes Asyncio coroutines” but seems a bit stale (latest commit around 2 years ago)

Well that didn’t age so well did it? Version 5.0 of Celery didn’t implement asyncio compatibility thus we cannot know when and if this will ever be implemented… Leaving this here for response legacy reasons (as it was the answer at the time) and for comment continuation.

That will be possible from Celery version 5.0 as stated on the official site:

http://docs.celeryproject.org/en/4.0/whatsnew-4.0.html#preface

  1. The next major version of Celery will support Python 3.5 only, where we are planning to take advantage of the new asyncio library.
  2. Dropping support for Python 2 will enable us to remove massive amounts of compatibility code, and going with Python 3.5 allows us to take advantage of typing, async/await, asyncio, and similar concepts there’s no alternative for in older versions.

The above was quoted from the previous link.

So the best thing to do is wait for version 5.0 to be distributed!

In the meantime, happy coding 🙂

More Related Contents:

  1. How to limit concurrency with Python asyncio?
  2. Simplest async/await example possible in Python
  3. How can I wrap a synchronous function in an async coroutine?
  4. asyncio.sleep() vs time.sleep() in Python
  5. how to add a coroutine to a running asyncio loop?
  6. How does asyncio actually work?
  7. Asynchronous method call in Python?
  8. How can I periodically execute a function with asyncio?
  9. Python 3.7 – asyncio.sleep() and time.sleep()
  10. FastAPI runs api-calls in serial instead of parallel fashion
  11. asyncio.sleep() vs time.sleep()
  12. multiprocessing vs multithreading vs asyncio in Python 3
  13. How to use asyncio with existing blocking library?
  14. Asyncio.gather vs asyncio.wait
  15. Sharing python objects across multiple workers
  16. Combine awaitables like Promise.all
  17. What kind of problems (if any) would there be combining asyncio with multiprocessing?
  18. How to test Python 3.4 asyncio code?
  19. Does asyncio supports asynchronous I/O for file operations?
  20. What does “SSLError: [SSL] PEM lib (_ssl.c:2532)” mean using the Python ssl library?
  21. How to call an async function without await?
  22. multiprocessing vs multithreading vs asyncio
  23. Use asyncio and Tkinter (or another GUI lib) together without freezing the GUI
  24. Using asyncio.Queue for producer-consumer flow
  25. asyncio.ensure_future vs. BaseEventLoop.create_task vs. simple coroutine?
  26. How to use await in a python lambda
  27. How to set class attribute with await in __init__
  28. How can I await inside future-like object’s __await__?
  29. Is there a way to use asyncio.Queue in multiple threads?
  30. “asyncio.run() cannot be called from a running event loop” when using Jupyter Notebook

Leave a Comment