Why Async Code Breaks Inside Celery Tasks (FastAPI + Celery)

Why Async Code Breaks Inside Celery Tasks (FastAPI + Celery)

When you first pair FastAPI with Celery, everything feels smooth:

• FastAPI handles asynchronous APIs
• Celery handles heavy background jobs
• Redis or RabbitMQ handles queueing

Then one day, you write something like this inside a Celery task:

@app.task
def process_report(user_id):
    data = await fetch_user_data(user_id)  # ❌ Problem!

Boom.

You get an error like:

RuntimeError: This event loop is already running

Or worse, the task hangs indefinitely.

So why does this happen, and what’s the clean way to solve it?

---

Why Async Code Fails Inside Celery Tasks

Celery workers are not asynchronous by default.

They execute tasks in separate worker processes and do not natively understand Python's async/await model.

When a Celery task tries to run:

await send_email()
await fetch_data()
await some_async_call()

Celery doesn't know how to manage the coroutine or event loop properly.

This often leads to:

• RuntimeError: This event loop is already running
• Event loop conflicts
• Hanging tasks
• Unpredictable behavior in production

Some developers attempt to work around this using:

asyncio.run(fetch_user_data(user_id))

While this may appear to work initially, it creates a new event loop for every task execution, which can become inefficient and problematic under load.

---

The Better Pattern: Let FastAPI Handle Async Work

Instead of executing asynchronous code directly inside Celery, delegate the async work back to FastAPI.

Principle

• Celery remains responsible for orchestration and background scheduling.
• FastAPI remains responsible for asynchronous operations.

The Celery worker simply triggers an internal FastAPI endpoint that executes the async logic safely.

---

Architecture

Client
   │
   ▼
FastAPI
   │
   ▼
Celery Task
   │
   ▼
Internal FastAPI Endpoint
   │
   ▼
Async Function

Flow:

Celery Task
      │
      ▼
HTTP Request
      │
      ▼
FastAPI Endpoint
      │
      ▼
await fetch_user_data()

---

Implementation

1. Internal FastAPI Endpoint

from fastapi import APIRouter
router = APIRouter()

@router.post("/internal/fetch-user")
async def fetch_user_endpoint(payload: dict):
    user_id = payload["user_id"]
    data= await fetch_user_data(user_id)
    return data

---

2. Synchronous Trigger Function

import requests

def trigger_user_processing(user_id):
    url = "http://localhost:8000/internal/fetch-user"

    response = requests.post(
        url,
        json={"user_id": user_id}
    )

    return response.json()

---

3. Celery Task

@app.task
def process_data(metadata):
    user_id = metadata.get("user_id")
    data = trigger_user_processing(user_id)

    # Additional processing
    # Update database
    # Send notifications
    # Generate reports

---

What Happened?

The Celery worker performs a normal synchronous HTTP request:

Celery → requests.post(...)

FastAPI receives the request and executes the async logic:

await fetch_user_data(user_id)

Since FastAPI is already running on an ASGI server such as Uvicorn, it has a proper event loop available.

No event loop conflicts occur.

---

Why This Works

Celery Stays Synchronous

No coroutine management is required inside the worker.

FastAPI Handles Async Operations

The event loop is managed by Uvicorn/ASGI, which is designed for asynchronous execution.

Clear Separation of Responsibilities

ComponentResponsibilityFastAPIAsync I/O operationsCeleryBackground orchestrationRedis/RabbitMQTask queueingWorkersLong-running processing

Easier to Scale

• FastAPI instances can scale independently.
• Celery workers can scale independently.
• Async code remains centralized.

---

Considerations

While this pattern is simple and effective, it introduces an HTTP hop between Celery and FastAPI.

For high-throughput systems, alternative approaches may be preferable:

• Using async-native task queues
• Running async code through a dedicated service layer
• Leveraging Celery's evolving async support (depending on version and requirements)

However, for many microservice architectures, an internal endpoint provides a clean and reliable solution.

---

Final Thoughts

If you're using FastAPI and Celery and encountering async/await issues:

• Don't force async code directly into Celery tasks.
• Don't rely heavily on asyncio.run() inside workers.
• Let FastAPI own asynchronous operations.
• Let Celery focus on orchestration and background execution.

A simple internal API call can eliminate event loop conflicts and keep your architecture predictable, maintainable, and production-friendly.