Skip to content

Calling Async Code from Celery Tasks

Celery tasks are synchronous functions, and the async half of a codebase does not care: the HTTP client is httpx.AsyncClient, the database driver is asyncpg, the internal helpers are all async def. The obvious bridge is asyncio.run(coro()) inside the task, and it works — but it creates and destroys an event loop per task, and with it every connection pool, client and cached resource the coroutine touches. Measured on a trivial SELECT 1 against PostgreSQL, that costs 14.75 ms per call against 0.21 ms with a loop and pool that survive between tasks: a 71x difference, almost none of which is the loop itself (0.049 ms).

Prerequisites

What a new event loop per task really costs 3 bars comparing asyncio.run + connect per task with the others. What a new event loop per task really costs asyncio.run + connect per task 14.75 ms shared loop + pool 0.21 ms loop creation alone 0.049 ms Same query (SELECT 1) against the same PostgreSQL, 100 calls each. The loop is cheap; the connections and pools that cannot survive it are not.

1. Start with asyncio.run, and know what it costs

@app.task(name="fetch_report")
def fetch_report(report_id: int) -> dict:
    return asyncio.run(build_report(report_id))        # a fresh loop every time

This is correct, simple and fine when tasks are infrequent or the coroutine has no expensive setup. asyncio.run creates a loop, runs the coroutine, cancels leftover tasks, shuts down async generators and closes the loop — about 0.049 ms of pure overhead.

The cost is everything the loop owned. A pool created inside the coroutine is built and torn down per task; an httpx.AsyncClient created there does a fresh TLS handshake; a cached authentication token is discarded. On the measured benchmark, connecting to PostgreSQL per call was what turned 0.21 ms into 14.75 ms.

One hard rule: asyncio.run fails with RuntimeError: asyncio.run() cannot be called from a running event loop. That happens if the task is invoked from async code — in an eager-mode test, for example — so tasks written this way are not callable from a coroutine.

Verify: time a task that touches a database or HTTP service; if it is dominated by connection setup, the loop is being recreated.

2. Run one loop per worker process

When most tasks need async code, give the worker process a loop that outlives individual tasks:

_loop: asyncio.AbstractEventLoop | None = None
_resources: dict = {}


def _ensure_loop() -> asyncio.AbstractEventLoop:
    global _loop
    if _loop is None:
        _loop = asyncio.new_event_loop()
        threading.Thread(target=_loop.run_forever, daemon=True, name="async-bridge").start()

        async def build():                             # built INSIDE the loop's thread
            _resources["pool"] = await asyncpg.create_pool(DSN, min_size=2, max_size=8)
            _resources["http"] = httpx.AsyncClient(timeout=10)

        asyncio.run_coroutine_threadsafe(build(), _loop).result(timeout=30)
    return _loop


def run_async(coro, timeout: float = 60):
    return asyncio.run_coroutine_threadsafe(coro, _ensure_loop()).result(timeout=timeout)


@app.task(name="fetch_report")
def fetch_report(report_id: int) -> dict:
    return run_async(build_report(report_id, _resources))

run_coroutine_threadsafe is the only safe way to submit work to a loop running in another thread, and .result(timeout=...) blocks the Celery worker thread until it finishes — which is what a synchronous task wants.

The build() indirection is not stylistic. Constructing a pool in the main thread fails immediately:

RuntimeError: There is no current event loop in thread 'MainThread'.

because asyncpg.create_pool() captures a loop at construction time. Anything loop-bound must be created inside a coroutine that runs on the bridge loop.

Verify: the median per-task time drops to the query's own cost, and the database shows a stable connection count rather than churn.

The shared-loop pattern, correctly 5 ordered steps. The shared-loop pattern, correctly create the loop lazily per worker process, not per task run_forever in a daemon thread it outlives every task build pools inside that thread not in the main thread run_coroutine_threadsafe(...).result() with a timeout close it on worker_shutdown a Celery signal Pools built in the main thread bind to the wrong loop and fail at the first query.

3. Respect the process model

Celery's default prefork pool means each worker is a separate process, forked from a parent. Two consequences:

  • Each process needs its own loop and pools. In testing, four worker processes produced four distinct loop objects — correct, and worth remembering when sizing pools: max_size=8 with 8 workers is 64 connections.
  • Nothing loop-bound may be created before the fork. A pool created at import time exists in the parent and is inherited, half-broken, by every child. The lazy _ensure_loop() above avoids this by construction, and Celery's worker_process_init signal is the explicit hook if you prefer eager setup.
@worker_process_init.connect
def init_bridge(**_):
    _ensure_loop()


@worker_shutdown.connect
def close_bridge(**_):
    if _loop is not None:
        asyncio.run_coroutine_threadsafe(_close_resources(), _loop).result(timeout=10)
        _loop.call_soon_threadsafe(_loop.stop)

Closing the resources matters for the same reason it does in an ASGI lifespan: a worker that exits without closing its pool leaves connections for the database to time out.

Verify: the database's connection count equals pool_size × worker_processes and returns to zero when the workers stop.

4. Keep task-level timeouts on both sides

A synchronous task blocked in .result() is invisible to asyncio's cancellation, so the timeout must be explicit — and it should be smaller than Celery's own limits:

def run_async(coro, timeout: float = 60):
    future = asyncio.run_coroutine_threadsafe(coro, _ensure_loop())
    try:
        return future.result(timeout=timeout)
    except concurrent.futures.TimeoutError:
        future.cancel()                                # ask the coroutine to stop
        raise

future.cancel() on a concurrent.futures.Future from run_coroutine_threadsafe does propagate into the coroutine — unlike a thread running arbitrary blocking code, as described in timing out blocking calls in threads. That makes the bridge better behaved than most, provided the coroutine is cancellation-safe.

Celery's soft_time_limit raises inside the worker thread while it is blocked in .result(), so pair the two: the bridge timeout slightly below the soft limit, and the soft limit below the hard one.

Verify: a coroutine that hangs is abandoned at the bridge timeout, and the worker is free for the next task.

5. Consider not bridging at all

The bridge is a real amount of machinery — a thread, a lazily built loop, per-process resources, two layers of timeout, and a shutdown hook. If most of your tasks are async, an async-native queue removes all of it:

@broker.task                                           # taskiq
async def fetch_report(report_id: int) -> dict:
    return await build_report(report_id)

arq and taskiq run coroutine tasks directly on one loop per worker, with pools created once in the worker's startup hook. Choosing between Celery, arq and taskiq covers the trade — Celery's scheduling, routing and ecosystem against the simplicity of a worker that speaks your code's language natively.

The middle path, where a large Celery deployment exists and only some tasks are async, is the bridge above. Keep it in one module, with one run_async helper, so the eventual migration touches one file.

Verify: count the tasks that call run_async; if it is most of them, price the migration.

Four ways to run async code from a sync worker A grid of 4 rows by 2 columns. Four ways to run async code from a sync worker approach pools reused? when it fits asyncio.run per task no: everything rebuilt rare tasks, cheap clients shared loop in a thread yes, per worker process Celery you must keep arq or taskiq yes, natively new work, no Celery legacy call a separate async service yes, in that service the async part is substantial Measured: the first row cost 71 times the second for the same query.
How much async does this worker need? A decision on How much of the work is async with 3 outcomes. How much async does this worker need? How much of the work is async? an occasional call asyncio.run simple, and the cost is bounded most tasks, Celery required a shared loop pools survive between tasks most tasks, free choice arq or taskiq no bridge at all The bridge is worth its complexity only when pools and clients are actually reused.

Verification

The bridge is correct when:

  • One loop exists per worker process, created lazily after the fork.
  • Pools and clients are built inside the loop's thread and reused across tasks.
  • Per-task overhead is the work's own cost, not connection setup.
  • Timeouts exist on both sides, with the bridge below Celery's soft limit.
  • Resources close on worker_shutdown.
  • Pool sizes account for the worker count.

Pitfalls & edge cases

  • asyncio.run inside an already-running loop. Raises RuntimeError; tasks written that way cannot be called from async code.
  • Creating pools before the fork. Children inherit unusable objects; build lazily or in worker_process_init.
  • Constructing loop-bound objects in the main thread. asyncpg.create_pool() raises RuntimeError: There is no current event loop.
  • loop.run_until_complete from the task thread. The loop is already running in another thread; use run_coroutine_threadsafe.
  • Forgetting the .result() timeout. A hung coroutine holds the Celery worker forever.
  • Sharing one loop across threads without care. Only call_soon_threadsafe and run_coroutine_threadsafe are safe from outside.

Frequently Asked Questions

How do I call async functions from a Celery task?

The simple way is asyncio.run(coro()) inside the task. The efficient way, when many tasks need it, is one event loop per worker process running in a daemon thread, with coroutines submitted via asyncio.run_coroutine_threadsafe(coro, loop).result(timeout=...).

Is asyncio.run in a Celery task slow?

The loop itself costs about 0.049 ms. What is slow is everything the loop owned being rebuilt: measured on a trivial database query, connecting per task cost 14.75 ms against 0.21 ms with a shared loop and pool — 71 times more.

Why does asyncpg.create_pool fail inside my Celery worker?

Because it was constructed in the main thread, where no event loop exists, while the loop runs in another thread. Wrap the creation in an async function and submit that with run_coroutine_threadsafe, so the pool is built inside the loop's thread.

Can I share one event loop across Celery worker processes?

No. With the prefork pool each worker is a separate process with its own memory, so each needs its own loop and its own pools. Remember that pool sizes multiply: max_size=8 across 8 workers is 64 connections to the database.

Should I use Celery at all for async code?

If most tasks are coroutines and nothing ties you to Celery, arq or taskiq run them natively with one loop per worker and no bridge. Keep Celery when its scheduling, routing and ecosystem matter, and confine the bridge to a single module.