# FastAPI Task Dispatch with KEDA Autoscaling (/integrations/celery/scenarios/fastapi-tasks-kubernetes)



## Scenario [#scenario]

A FastAPI web tier accepts HTTP requests and offloads slow work to background tasks. Each request dispatches a Celery task with `task.delay(...)` and immediately returns a task id, so the client never blocks on the work itself. A separate endpoint lets the client poll task status and fetch the result once it is ready.

The twist: there is **no Redis and no RabbitMQ** anywhere in the stack. Both the Celery broker and the result backend are KubeMQ, reached through the `kubemq://` transport. The web pods and the worker pods talk to a single KubeMQ broker, and a KEDA `ScaledObject` watches the queue depth on that broker to scale the worker Deployment up during request bursts and back down when the queue drains.

This scenario walks the full path: a shared Celery app, the FastAPI endpoints, an optional WebSocket progress stream, the worker tuning profile for API workloads, containerization with `docker-compose` for local runs, the Kubernetes Deployment, the KEDA trigger, and the reliability settings that keep fast API tasks honest.

## Architecture [#architecture]

The web tier and the workers are separate Deployments that share one KubeMQ broker. FastAPI pods only ever produce tasks and read results; the worker Deployment consumes tasks, and KEDA scales it from the queue depth that KubeMQ reports.

<Mermaid
  chart="flowchart LR
    Client[&#x22;HTTP Client&#x22;] -->|&#x22;POST /tasks&#x22;| FA[&#x22;FastAPI Pods<br/>(web tier)&#x22;]
    FA -->|&#x22;task.delay() via kubemq://&#x22;| KMQ[&#x22;KubeMQ Broker<br/>Queues&#x22;]
    KMQ -->|&#x22;consume tasks&#x22;| W[&#x22;Celery Worker<br/>Deployment&#x22;]
    W -->|&#x22;store result&#x22;| RB[&#x22;Result Backend<br/>(queue peek)&#x22;]
    FA -->|&#x22;AsyncResult poll&#x22;| RB
    KMQ -.->|&#x22;queueLength&#x22;| KEDA[&#x22;KEDA ScaledObject&#x22;]
    KEDA -.->|&#x22;scale replicas&#x22;| W"
/>

## Shared Celery app [#shared-celery-app]

The web tier and the workers import the **same** Celery app module. Importing `kubemq_celery` registers the `kubemq://` transport with Kombu, and both the broker and the result backend are read from environment variables so the identical image runs locally and in-cluster. This is the app from `examples/integrations/fastapi_integration.py`.

```python title="tasks.py"
from __future__ import annotations

import os
import time
from typing import Any

from celery import Celery

import kubemq_celery  # noqa: F401  registers the kubemq:// transport

celery_app = Celery(
    "fastapi_integration",
    broker=os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
    result_backend=os.environ.get("CELERY_RESULT_BACKEND", "kubemq://localhost:50000"),
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    result_expires=3600,
    task_track_started=True,
    worker_prefetch_multiplier=1,
)


@celery_app.task(bind=True)
def process_data(self, data: str, options: dict[str, Any] | None = None) -> dict:
    """Process data with configurable transformations."""
    options = options or {}
    self.update_state(state="PROGRESS", meta={"step": "processing", "percent": 50})
    time.sleep(1)

    result = data
    if options.get("uppercase"):
        result = result.upper()
    if options.get("reverse"):
        result = result[::-1]

    return {
        "input": data,
        "output": result,
        "options_applied": list(options.keys()),
    }
```

<Callout type="info">
  `task_track_started=True` makes Celery emit a `STARTED` state the moment a worker picks up the task, which gives the status endpoint a useful intermediate state between `PENDING` and `SUCCESS`. `import kubemq_celery` must run before the app connects, or Kombu will not recognize the `kubemq://` scheme.
</Callout>

## FastAPI endpoints [#fastapi-endpoints]

FastAPI is the producer. The `POST` endpoint dispatches the task and returns its id without waiting; the `GET` endpoint builds an `AsyncResult` and maps the Celery state onto a response model. This is `examples/integrations/fastapi_integration.py`.

```python title="api.py"
from typing import Any

from celery.result import AsyncResult
from fastapi import FastAPI
from pydantic import BaseModel

from tasks import celery_app, process_data


class TaskRequest(BaseModel):
    data: str
    options: dict[str, Any] | None = None


class TaskResponse(BaseModel):
    task_id: str
    status: str
    detail: str


class TaskStatus(BaseModel):
    task_id: str
    status: str
    result: Any | None = None
    error: str | None = None
    progress: dict[str, Any] | None = None


api = FastAPI(
    title="KubeMQ Celery API",
    description="FastAPI + Celery with KubeMQ transport",
    version="1.0.0",
)


@api.post("/tasks/process", response_model=TaskResponse)
async def submit_task(request: TaskRequest) -> TaskResponse:
    """Submit a data processing task to the KubeMQ broker."""
    result = process_data.delay(request.data, request.options)
    return TaskResponse(
        task_id=result.id,
        status="PENDING",
        detail="Task submitted to KubeMQ broker",
    )


@api.get("/tasks/{task_id}", response_model=TaskStatus)
async def get_task_status(task_id: str) -> TaskStatus:
    """Check the status of a submitted task."""
    result = AsyncResult(task_id, app=celery_app)

    if result.state == "PROGRESS":
        return TaskStatus(task_id=task_id, status="PROGRESS", progress=result.info)
    elif result.state == "SUCCESS":
        return TaskStatus(task_id=task_id, status="SUCCESS", result=result.result)
    elif result.state == "FAILURE":
        return TaskStatus(task_id=task_id, status="FAILURE", error=str(result.result))
    else:
        return TaskStatus(task_id=task_id, status=result.state)
```

The dispatch-and-poll flow looks like this from the client side:

```bash title="dispatch and poll"
# Submit a task -> returns a task id
curl -X POST http://localhost:8000/tasks/process \
  -H "Content-Type: application/json" \
  -d '{"data": "hello world", "options": {"uppercase": true}}'
# {"task_id": "a1b2c3...", "status": "PENDING", "detail": "Task submitted to KubeMQ broker"}

# Poll status with the returned id
curl http://localhost:8000/tasks/a1b2c3...
# {"task_id": "a1b2c3...", "status": "SUCCESS", "result": {"input": "hello world", "output": "HELLO WORLD", ...}}
```

## Real-time progress [#real-time-progress]

Polling is fine for short tasks, but a longer task can stream its progress to the browser over a WebSocket instead. The task reports fine-grained progress with `self.update_state(...)`, and the WebSocket endpoint reads those `PROGRESS` states from the result backend and forwards them to the client until the task reaches a terminal state. This combines `examples/integrations/fastapi_websocket_progress.py` with the progress task from `examples/monitoring/progress_tracking.py`.

```python title="ws_progress.py"
import asyncio

from celery.result import AsyncResult
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

from tasks import celery_app


@celery_app.task(bind=True)
def long_running_task(self, steps: int = 10) -> dict:
    """Task that reports progress at each step."""
    import time

    for i in range(1, steps + 1):
        self.update_state(
            state="PROGRESS",
            meta={
                "current": i,
                "total": steps,
                "percent": int(i / steps * 100),
                "message": f"Processing step {i}/{steps}",
            },
        )
        time.sleep(1)
    return {"steps_completed": steps, "status": "done"}


api = FastAPI()


@api.post("/tasks/process")
async def submit_task(steps: int = 5) -> dict:
    result = long_running_task.delay(steps)
    return {"task_id": result.id, "status": "PENDING", "ws_url": f"/ws/progress/{result.id}"}


@api.websocket("/ws/progress/{task_id}")
async def websocket_progress(websocket: WebSocket, task_id: str):
    """Stream task progress updates over WebSocket."""
    await websocket.accept()
    max_iterations = 300
    try:
        for _ in range(max_iterations):
            result = AsyncResult(task_id, app=celery_app)

            if result.state == "PROGRESS":
                await websocket.send_json({"status": "PROGRESS", "progress": result.info})
            elif result.state == "SUCCESS":
                await websocket.send_json({"status": "SUCCESS", "result": result.result})
                break
            elif result.state == "FAILURE":
                await websocket.send_json({"status": "FAILURE", "error": str(result.result)})
                break
            else:
                await websocket.send_json({"status": result.state})

            await asyncio.sleep(0.5)
    except WebSocketDisconnect:
        pass
```

<Callout type="info">
  Each `update_state()` overwrites the task's entry on the KubeMQ result channel, and the client reads the latest value by peeking that channel — the backend does not keep a history of progress snapshots. The endpoint caps its read loop at `max_iterations` so a never-finishing task cannot hold the socket open forever.
</Callout>

## Worker config for API workload [#worker-config-for-api-workload]

Tasks dispatched from web requests need to start fast, not batch efficiently. The low-latency profile from `docs/performance.md` raises concurrency so more requests run in parallel, keeps `prefetch_multiplier=1` so no worker hoards queued tasks while it is busy, and sets `max_batch_size=1` to pull one message at a time for the lowest possible dispatch latency.

```python title="worker_config.py"
celery_app.conf.update(
    worker_concurrency=8,
    worker_prefetch_multiplier=1,
    broker_transport_options={
        "wait_timeout": 1,
        "max_batch_size": 1,  # minimize latency
    },
)
```

<Callout type="warn">
  Keep `wait_timeout` below Celery's `drain_events` timeout (default 2s) to avoid transport deadlocks. `wait_timeout: 1` satisfies this.
</Callout>

## Containerize and run locally [#containerize-and-run-locally]

Package the shared app into one image and run the same image as both the web tier and the worker, differing only by command. The `Dockerfile` from `examples/kubernetes/Dockerfile` installs dependencies with `uv`.

```dockerfile title="Dockerfile"
FROM python:3.12-slim

WORKDIR /app

# Install uv and dependencies
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
COPY pyproject.toml .
RUN uv pip install --system --no-cache .

# Copy application code
COPY src/ src/
COPY examples/ examples/

ENV PYTHONPATH=/app/src:/app/examples

# Health check
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD python -c "import kubemq_celery; print('ok')" || exit 1

CMD ["celery", "-A", "basic_task", "worker", "--loglevel=info"]
```

Before touching Kubernetes, validate the whole stack with `docker-compose`. This compose file (adapted from `examples/kubernetes/docker-compose.yaml`) brings up a KubeMQ broker with a health check and a Celery worker that waits for the broker to be ready.

```yaml title="docker-compose.yaml"
services:
  kubemq:
    image: europe-docker.pkg.dev/kubemq/images/kubemq:next
    ports:
      - "50000:50000"  # gRPC
      - "9090:9090"    # Shared HTTP server (REST/health)
      - "8080:8080"    # Management API dashboard
    environment:
      - KUBEMQ_LOG_LEVEL=info
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:9090/health"]
      interval: 10s
      timeout: 5s
      retries: 3
      start_period: 10s

  worker:
    build:
      context: .
      dockerfile: Dockerfile
    command: celery -A tasks worker --loglevel=info -Q celery --concurrency=8
    environment:
      - CELERY_BROKER_URL=kubemq://kubemq:50000
      - CELERY_RESULT_BACKEND=kubemq://kubemq:50000
    depends_on:
      kubemq:
        condition: service_healthy
    restart: unless-stopped

  api:
    build:
      context: .
      dockerfile: Dockerfile
    command: uvicorn api:api --host 0.0.0.0 --port 8000
    ports:
      - "8000:8000"
    environment:
      - CELERY_BROKER_URL=kubemq://kubemq:50000
      - CELERY_RESULT_BACKEND=kubemq://kubemq:50000
    depends_on:
      kubemq:
        condition: service_healthy
    restart: unless-stopped
```

```bash title="run locally"
docker compose up -d
# KubeMQ dashboard: http://localhost:8080  (Management API — watch queue depth here)
# FastAPI:          http://localhost:8000/docs
```

## Deploy to Kubernetes [#deploy-to-kubernetes]

In-cluster, the broker runs as a StatefulSet and the workers as a separate Deployment. Workers reach the broker through its Kubernetes service DNS name, and the readiness and liveness probes use `celery inspect ping`, which exercises the transport's `verify_connection()` against the broker. This worker Deployment is from `docs/kubernetes.md`.

```yaml title="worker-deployment.yaml"
apiVersion: apps/v1
kind: Deployment
metadata:
  name: celery-worker
  namespace: default
  labels:
    app: celery-worker
spec:
  replicas: 3
  selector:
    matchLabels:
      app: celery-worker
  template:
    metadata:
      labels:
        app: celery-worker
    spec:
      containers:
        - name: worker
          image: myapp:latest
          command:
            - celery
            - -A
            - tasks
            - worker
            - --loglevel=info
            - --concurrency=8
          env:
            - name: CELERY_BROKER_URL
              value: "kubemq://kubemq.default.svc:50000"
            - name: CELERY_RESULT_BACKEND
              value: "kubemq://kubemq.default.svc:50000"
          resources:
            requests:
              cpu: 250m
              memory: 256Mi
            limits:
              cpu: 1000m
              memory: 512Mi
          readinessProbe:
            exec:
              command:
                - celery
                - -A
                - tasks
                - inspect
                - ping
                - --timeout=5
            initialDelaySeconds: 30
            periodSeconds: 30
            timeoutSeconds: 10
          livenessProbe:
            exec:
              command:
                - celery
                - -A
                - tasks
                - inspect
                - ping
                - --timeout=5
            initialDelaySeconds: 60
            periodSeconds: 60
            timeoutSeconds: 10
```

The broker itself runs as a StatefulSet exposing gRPC on `50000`, the shared HTTP server (REST/health) on `9090`, and the Management API dashboard on `8080`. Inside the cluster, workers and web pods address it as `kubemq.default.svc:50000`.

```yaml title="kubemq-statefulset.yaml"
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kubemq
  namespace: default
spec:
  serviceName: kubemq
  replicas: 3  # HA cluster
  selector:
    matchLabels:
      app: kubemq
  template:
    metadata:
      labels:
        app: kubemq
    spec:
      containers:
        - name: kubemq
          image: europe-docker.pkg.dev/kubemq/images/kubemq:next
          ports:
            - containerPort: 50000
              name: grpc
            - containerPort: 9090
              name: http
            - containerPort: 8080
              name: dashboard
          readinessProbe:
            tcpSocket:
              port: 50000
            initialDelaySeconds: 5
            periodSeconds: 10
          livenessProbe:
            tcpSocket:
              port: 50000
            initialDelaySeconds: 15
            periodSeconds: 20
```

<Callout type="warn">
  The worker probes use `celery inspect ping`, which connects through the broker. Give the broker StatefulSet a generous `initialDelaySeconds` (the worker readiness probe waits 30s) so the broker is reachable before workers start failing health checks during a cold start.
</Callout>

## Autoscale with KEDA [#autoscale-with-keda]

HPA scales on CPU and memory, which lag behind a queue that is filling from a request burst. KEDA scales on the metric that actually matters here — the KubeMQ queue depth. The `kubemq` trigger watches the `celery` channel and adds workers when more than 10 messages are pending, scaling between 1 and 20 replicas. This is `examples/kubernetes/keda-scaler.yaml`.

```yaml title="keda-scaler.yaml"
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: celery-worker-scaler
  namespace: default
  labels:
    app: celery-worker
    component: autoscaling
spec:
  scaleTargetRef:
    name: celery-worker          # must match Deployment name
  minReplicaCount: 1             # always keep at least 1 worker
  maxReplicaCount: 20            # maximum workers
  pollingInterval: 10            # check queue depth every 10 seconds
  cooldownPeriod: 60             # wait 60s before scaling down
  triggers:
    - type: kubemq
      metadata:
        # KubeMQ broker address (cluster-internal)
        address: "kubemq.default.svc:50000"
        # Queue channel to monitor (Celery default queue)
        channel: "celery"
        # Scale threshold: add workers when queue depth > 10
        queueLength: "10"
```

Install KEDA and apply the scaler:

```bash title="install KEDA"
helm repo add kedacore https://kedacore.github.io/charts
helm install keda kedacore/keda --namespace keda --create-namespace
kubectl apply -f keda-scaler.yaml
```

When a request burst hits the FastAPI tier, the `celery` channel depth climbs, KEDA notices within `pollingInterval` (10s) and scales the worker Deployment up; once the queue drains it scales back down after `cooldownPeriod` (60s). Watch the depth live in the KubeMQ Management API dashboard at port `8080` (or query the REST `/queue/info` endpoint on the shared HTTP server at `9090`).

## Reliability notes [#reliability-notes]

A few defaults matter for fast, request-driven tasks:

* **`acks_early` is the right default.** The message is acknowledged before the task runs, so a quick API task is not re-delivered if a worker dies mid-execution. For at-least-once delivery on a critical task, opt in per task with `acks_late=True` and `reject_on_worker_lost=True` — but be aware that KubeMQ's transaction timeout can expire for long-running tasks (> 60s), so use `acks_early` or make the task idempotent for those.

  ```python title="reliability.py"
  celery_app.conf.task_acks_late = False  # default — good for fast API tasks

  @celery_app.task(acks_late=True, reject_on_worker_lost=True)
  def critical_payment(payment_id: str) -> dict:
      """Process payment — must not be lost. Idempotent via payment_id dedup."""
      ...
  ```

* **Configure a dead letter queue for poison messages.** The DLQ operates at the broker level (receive count) and is independent of Celery's task-level `max_retries`; the two work together so a message that repeatedly fails to process lands in a DLQ channel instead of looping forever.

* **Keep `result_expires` at or below 86400.** For a polling API, results only need to live long enough for the client to fetch them. The shared app sets `result_expires=3600` (one hour); the documented maximum is 86400 (24 hours). Bounding this keeps the result backend from accumulating stale entries.

## Next steps [#next-steps]

<Cards>
  <Card title="Kubernetes Deployment" href="/integrations/celery/how-to/kubernetes" description="Full deployment guide — broker StatefulSet, worker resources, service discovery, health checks, and monitoring." />

  <Card title="Error Handling" href="/integrations/celery/how-to/error-handling" description="Retries, dead letter queues, and acks_late vs acks_early on KubeMQ." />

  <Card title="API Reference" href="/integrations/celery/reference/configuration" description="URL schemes, transport options, and result backend reference." />
</Cards>