Flask Async Celery

Run async def Celery tasks on a persistent asyncio event loop with bounded concurrency, Flask integration, and Celery consumer backpressure.

Features

• Persistent asyncio event loop in a dedicated thread per Celery worker process.
• Run native async def Celery tasks.
• Bounded asynchronous concurrency with max_tasks.
• Celery request context propagation into the asyncio execution thread.
• self.retry() support for async tasks.
• Normal synchronous Celery tasks continue to work.
• Redis consumer-side backpressure through Celery's worker_disable_prefetch.
• Flask extension with simple configuration.
• Graceful asyncio executor shutdown.
• Compatible with Celery 5.6.x and Python 3.10+.

Architecture

                         Redis
                           │
                           ▼
                    Celery Consumer
                           │
                 worker_disable_prefetch
                           │
                           ▼
                     AsyncIOPool
                    max_tasks = N
                           │
                           ▼
                    AsyncExecutor
                  asyncio.Semaphore(N)
                           │
                    Persistent loop
                           │
              ┌────────────┼────────────┐
              ▼            ▼            ▼
          Async task    Async task    Async task

The package separates Celery's worker execution from asyncio execution:

1. Celery receives and traces the task.
2. AsyncIOPool bridges Celery execution into the asyncio executor.
3. AsyncExecutor owns a persistent asyncio event loop.
4. A semaphore limits active async tasks.
5. Celery's Redis worker_disable_prefetch option can prevent the consumer from reserving work beyond the available pool capacity.

Requirements

• Python 3.10+
• Celery 5.6.x
• Flask 2.3+
• Redis when using the Redis broker/result backend and consumer backpressure.

Installation

From PyPI:

pip install flask-async-celery

For Redis support:

pip install "flask-async-celery[redis]"

For development and testing:

pip install "flask-async-celery[test]"

Basic Flask Setup

import asyncio

from flask import Flask

from flask_async_celery import AsyncCelery, AsyncTask


app = Flask(__name__)

celery = AsyncCelery(
    app,
    broker_url="redis://127.0.0.1:6379/0",
    result_backend="redis://127.0.0.1:6379/1",
    max_tasks=5,
    disable_prefetch=True,
)


@celery.task(base=AsyncTask)
async def my_task(value):
    await asyncio.sleep(1)
    return value * 2

Send the task normally:

result = my_task.delay(10)

print(result.get(timeout=30))
# 20

Flask Configuration

Configuration can be supplied through Flask:

app.config["ASYNC_CELERY_MAX_TASKS"] = 10
app.config["ASYNC_CELERY_DISABLE_PREFETCH"] = True

celery = AsyncCelery(
    app,
    broker_url="redis://127.0.0.1:6379/0",
    result_backend="redis://127.0.0.1:6379/1",
)

Available settings:

Setting	Default	Description
ASYNC_CELERY_MAX_TASKS	20	Maximum number of concurrently executing async tasks.
ASYNC_CELERY_DISABLE_PREFETCH	True	Enables Celery consumer-side backpressure.

Constructor arguments can also be used directly:

celery = AsyncCelery(
    app,
    max_tasks=10,
    disable_prefetch=True,
)

Flask configuration takes precedence over the constructor defaults when the extension is initialized.

Worker Configuration

Run the worker using the package's custom pool:

celery -A your_app.celery worker \
    -P flask_async_celery.pool:AsyncIOPool \
    -c 5 \
    --loglevel=INFO

For example:

celery -A your_app.celery worker \
    -P flask_async_celery.pool:AsyncIOPool \
    -c 10 \
    --loglevel=INFO

The -c value should match the desired async concurrency.

The custom pool exposes its configured concurrency through num_processes, allowing Celery's consumer to use the same capacity when consumer-side prefetch is disabled.

Concurrency

Set the maximum number of simultaneously executing async tasks with:

celery = AsyncCelery(
    app,
    max_tasks=5,
)

With:

max_tasks = 5

the asyncio executor allows at most five active coroutines at once.

Additional work waits for an available execution slot.

This is different from simply creating more threads. The package uses one persistent asyncio event loop and runs async coroutines concurrently on that loop.

Consumer Backpressure

For Redis, Celery 5.6 supports:

worker_disable_prefetch = True

The extension enables this by default:

celery = AsyncCelery(
    app,
    max_tasks=5,
    disable_prefetch=True,
)

This provides two levels of protection:

Celery Consumer
      │
      │ Don't reserve beyond available capacity
      ▼
 AsyncIOPool
      │
      │ max_tasks
      ▼
AsyncExecutor
      │
      │ semaphore
      ▼
asyncio tasks

You can disable the consumer-side behavior:

celery = AsyncCelery(
    app,
    max_tasks=5,
    disable_prefetch=False,
)

When disabled, the asyncio executor still enforces its own concurrency limit.

Redis Requirement

worker_disable_prefetch is intended for supported Redis worker configurations. If you use another broker, verify that your Celery version and broker transport support this feature before relying on consumer-side backpressure.

The executor-level concurrency limit remains independent of consumer prefetch behavior.

Async Tasks

Use AsyncTask as the base class for native async tasks:

from flask_async_celery import AsyncTask


@celery.task(base=AsyncTask)
async def fetch_data():
    await some_async_operation()
    return "done"

The task can use normal Celery task features:

@celery.task(
    base=AsyncTask,
    bind=True,
    max_retries=3,
)
async def process_item(self, item_id):
    try:
        return await process(item_id)
    except TemporaryError as exc:
        raise self.retry(
            exc=exc,
            countdown=5,
        )

The Celery request context is propagated from the Celery worker thread into the asyncio execution thread, so task information such as the task ID, retry count, delivery information, and retry context remains available.

Synchronous Tasks

Normal synchronous tasks can still be used:

@celery.task
def sync_task(value):
    return value * 2

The package does not require every task to be asynchronous.

Retries

Async self.retry() is supported:

@celery.task(
    base=AsyncTask,
    bind=True,
    max_retries=3,
)
async def retrying_task(self):
    if should_retry():
        raise self.retry(countdown=5)

    return "success"

The Celery task request is preserved when execution moves from the Celery worker thread to the asyncio event loop. This allows Celery retry metadata and delivery information to remain available to the async task.

Exceptions

Exceptions raised by an async task propagate through the Celery execution path:

@celery.task(base=AsyncTask)
async def failing_task():
    raise RuntimeError("something went wrong")

Celery remains responsible for task failure state, result handling, retry behavior, and worker-level task tracing.

Graceful Shutdown

The asyncio executor runs in a dedicated daemon thread.

When the pool stops, the executor:

1. Stops accepting new work.
2. Stops the asyncio event loop.
3. Cancels pending asyncio tasks.
4. Waits for the loop thread to terminate.
5. Closes the event loop.

Public API

The main public API is intentionally small:

from flask_async_celery import AsyncCelery, AsyncTask

AsyncCelery

Flask integration and Celery configuration.

AsyncCelery(
    app=None,
    *,
    celery=None,
    broker_url=None,
    result_backend=None,
    max_tasks=20,
    disable_prefetch=True,
)

AsyncTask

Base class for asynchronous Celery tasks:

@celery.task(base=AsyncTask)
async def my_task():
    ...

Development

Clone the repository and install the project in editable mode:

git clone <repository-url>
cd flask-async-celery

pip install -e ".[test]"

Run the test suite:

pytest -v

The test suite covers:

• asyncio executor concurrency
• AsyncIOPool execution
• async task exceptions
• async retries
• synchronous retries
• real Celery worker integration
• Redis consumer backpressure
• Flask extension configuration
• end-to-end Flask/Celery/async execution

Project Structure

flask-async-celery/
├── pyproject.toml
├── README.md
├── src/
│   └── flask_async_celery/
│       ├── __init__.py
│       ├── extension.py
│       ├── executor.py
│       ├── bootstep.py
│       ├── pool.py
│       └── task.py
└── test/
    ├── conftest.py
    ├── test_executor.py
    ├── test_tasks.py
    ├── test_backpressure.py
    ├── test_celery_pool.py
    ├── test_worker_integration.py
    ├── test_extension.py
    └── test_extension_integration.py

Design Notes

This package does not replace Celery's task tracing and lifecycle handling.

Celery remains responsible for:

• task delivery
• task acknowledgment
• retries
• result state
• task IDs
• worker lifecycle
• task tracing

The package provides the asyncio execution layer and integrates it with Celery's pool interface.

The asyncio event loop is persistent for the lifetime of the worker process rather than creating a new event loop for every task.

Why a Persistent Event Loop?

Creating a new event loop for every task adds unnecessary setup and teardown overhead.

Instead, each worker process owns one persistent asyncio event loop:

Celery Worker Process
│
├── Celery Consumer
├── Celery task execution
├── bridge threads
│
└── asyncio event loop thread
    ├── Task A
    ├── Task B
    └── Task C

Async tasks can therefore share the same event loop while still being bounded by max_tasks.

Why Request Propagation?

Celery's request context is associated with the worker execution context.

The asyncio event loop runs in a separate thread, so the package explicitly transfers the current Celery request into that execution context.

This preserves information required by features such as:

self.request.id
self.request.retries
self.request.delivery_info
self.retry()

The request is pushed before async execution and removed afterward.

Limitations

Broker-specific Backpressure

Consumer-side worker_disable_prefetch support depends on Celery and the broker transport.

The asyncio executor's own max_tasks limit remains the final execution boundary.

Worker Pool

The worker must use:

flask_async_celery.pool:AsyncIOPool

for the package's asyncio execution model.

One Event Loop Per Worker Process

Each worker process owns its own asyncio event loop and concurrency limit.

For example:

celery -A your_app.celery worker \
    -P flask_async_celery.pool:AsyncIOPool \
    -c 5

creates a worker configuration with five execution slots.

If you run multiple worker processes, each process has its own pool and event loop.

Testing

Run the complete test suite:

pytest -v

The project currently tests the execution model with real Celery workers in addition to unit-level executor and pool tests.

A successful test run should show all tests passing.

Building the Package

Install the build tooling:

python -m pip install build

Build the source distribution and wheel:

python -m build

The generated files will be placed in:

dist/

You can inspect the generated wheel with:

python -m pip install dist/*.whl

Version

Current version:

0.1.0

License

Add the project's license information here.