Skip to main content

Celery pool to run coroutine tasks

Project description

Celery Pool AsyncIO

python version downloads format


  • Free software: Apache Software License 2.0


import asyncio
from celery import Celery

# celery_pool_asyncio importing is optional
# It imports when you run worker or beat if you define pool or scheduler
# but it does not imports when you open REPL or when you run web application.
# If you want to apply monkey patches anyway to make identical environment
# when you use REPL or run web application app it's good idea to import
# celery_pool_asyncio module
import celery_pool_asyncio  # noqa
# Sometimes noqa does not disable linter (Spyder IDE)

app = Celery()

    soft_time_limit=42,  # raises celery.exceptions.SoftTimeLimitExceeded inside the coroutine
    time_limit=300,  # breaks coroutine execution
async def my_task(self, *args, **kwargs):
    await asyncio.sleep(5)

async def my_simple_task(*args, **kwargs):
    await asyncio.sleep(5)

Then run celery:

$ celery worker -A -P celery_pool_asyncio:TaskPool

Monkey patching: wtf and why

There are many monkey patches should be applied to make application working, and some of them should be applied as early as possible. You are able to switch off any of them by setting environment variable CPA_MONKEY_DENY. Remember you should have a great reason to do it.

Except critical for work features it allows:

# await data sending to broker
async_result = await my_simple_task.delay()

# await wainting for AsyncResult
result = await async_result.get()

You can manually disable any of them by enumerating it comma separated:

$ env CPA_MONKEY_DENY=CELERY.SEND_TASK,ALL_BACKENDS celery worker -A -P celery_pool_asyncio:TaskPool

Disabling is available for:



Default scheduler doesn't work. PersistentScheduler is subclass of default celery scheduler.

Running celery with scheduler:

$ celery worker -A -P celery_pool_asyncio:TaskPool --scheduler celery_pool_asyncio:PersistentScheduler
$ celery beat -A --scheduler celery_pool_asyncio:PersistentScheduler

Embeding also supported:

$ celery worker -A -P celery_pool_asyncio:TaskPool --scheduler celery_pool_asyncio:PersistentScheduler -B

WARNING: embeded scheduler startup is not stable. It starts correctly in ~50% of cases. It looks like race condition. But after correct startup it works well. That's why it's good idea to run scheduler in separated process.

Celery Signals

from celery.signals import worker_init, worker_shutting_down

async def do_startup_async(sender, **kwargs):
    # Coroutine functions are available after pool initialized
    await MyClass.init_async()

def do_startup(sender, **kwargs):
    # regular functions are available too

async def do_shutdown(sender=None, **kwargs):
    await MyClass.shutdown()

More examples

There is an example project uses celery-pool-asyncio.



  • Allow to decorate corofunctions by celery.signals
  • Implement pool capacity (-c, --concurency)


  • Finalize monkey patcher refactoring. Now you able to switch off applying of any monkey patch. Remember with great power comes great responsibility
  • Implement soft_time_limit
  • Implement revoke
  • Fix keywords


  • Total monkey patching refactor. Now it is enabled by default, but you can manually disable some of features using environment variable CPA_MONKEY_DENY


  • Make Celery Beat working
    • Add async Celery Scheduler
    • More monkey patching
  • Move loop and loop_runner to own module
    • Avoid creating multiple loops and loop_runners per application


  • Large rework of await AsyncResult.get()
    • Works much better than earlier, but it's crap still
    • Added outnumber of monkey-patches
  • Fixed race condition on first microseconds of pool shutdown


  • Cleanup tracer, use namespase where it possible


  • Refactor monkey, split it
  • Move patch_send_task to own function
  • Add patch_result_get to await AsyncResult.get


  • Avoid building trace twice
  • Also this small performance optimization fixed AsyncResult.get


  • Fix graceful shutdown


  • Fix monkey: another function must be patched


  • Add changelog
  • Append documentation


  • Add monkey patcher to make brocker IO operations nonblocking


  • Refactor code
  • Fix found errors


  • Initial commit

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

celery-pool-asyncio-0.2.0.tar.gz (19.2 kB view hashes)

Uploaded source

Built Distribution

celery_pool_asyncio-0.2.0-py2.py3-none-any.whl (19.7 kB view hashes)

Uploaded py2 py3

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Fastly Fastly CDN Google Google Object Storage and Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page