Let’s imagine a very common problem. A user clicks a button in our app to generate a report, process a video, run an AI model, or exporting data. Whatever the task is, it takes time, seconds, maybe minutes. If our API tries to do that work synchronously, the request might time out, the server threads get blocked and the system doesn’t scale. In this post I am going to explain how to turn time consuming API requests into scalable background jobs using message broker and worker services.
In brief:

The user submits a request.Our system accepts it immediately, hands the work to background workers, and notifies the user when the result is ready.

That’s exactly what our architecture does. And most importantly, it is built to survive crashes, restarts, and heavy traffic.

Let’s walk through how it works. Also you can find the complete working project on my GitHub: async_jobs_pipeline_example

The Story of One Request

A user opens the UI and clicks Start Job. The browser sends a request to our API:

POST /jobs

Instead of doing the work itself, the API creates a job record in PostgreSQL and pushes a message to RabbitMQ. This message represents the task that needs to be done. The API responds immediately:

{
  "job_id": "abc123",
  "status": "queued"
}

The user waits for the task to be completed and the result be sent to him. But this waiting doesn’t block the server-client. The server simply sends notification to client as soon as the result gets ready.

RabbitMQ For Inter-Services Communications

RabbitMQ acts as a traffic controller between services. Instead of services calling each other directly, they communicate through queues. That gives us several benefits:

• services stay decoupled
• workers can scale independently
• failures don’t break the whole system
• jobs wait safely in queues

In our architecture we have three queues:

job_queue        -> worker1 listens to it and will write to worker2_queue
worker2_queue    -> worker2 listens to it notification_queue
notification_queue -> notification service listens to it

Each stage consumes from one queue and produces to the next. Think of it like a conveyor belt in a factory.

The API: Accepting Jobs

Our API is intentionally simple. Its job is to accept requests and enqueue work.

# api/main.py

@app.post("/jobs")
def create_job():
    job_id = str(uuid.uuid4())

    cur.execute(
        "INSERT INTO jobs (id, status) VALUES (%s, %s)",
        (job_id, "queued")
    )
    conn.commit()

    message = json.dumps({"job_id": job_id})

    channel.basic_publish(
        exchange="",
        routing_key="job_queue",
        body=message,
        properties=pika.BasicProperties(
            delivery_mode=2  # persistent message
        ),
    )

    return {"job_id": job_id}

Two important details here matter a lot. First, the message is marked as persistent:

delivery_mode=2

That tells RabbitMQ to store it on disk. So even if RabbitMQ crashes or restarts, the message won’t disappear. Second, the queue itself is durable.

channel.queue_declare(queue="job_queue", durable=True)

Having durable queue + persistent messages, means we will have no lost jobs.

Worker 1: The First Stage

Worker1 consumes jobs from the first queue. Its job is simple: take the task and produce an intermediate result.

from http import client
import json
import time
import pika
import psycopg2

def callback(ch, method, properties, body):
    data = json.loads(body)
    job_id = data["job_id"]

    print("Worker1 processing", job_id)

    time.sleep(3)

    result = f"processed-{job_id}"

    cur.execute(
        "UPDATE jobs SET result1=%s WHERE id=%s",
        (result, job_id)
    )
    conn.commit()

    message = json.dumps({
        "job_id": job_id,
        "result1": result
    })

    ch.basic_publish(
        exchange="",
        routing_key="worker2_queue",
        body=message,
        properties=pika.BasicProperties(delivery_mode=2),
    )

    ch.basic_ack(delivery_tag=method.delivery_tag)

def callback(ch, method, properties, body):
    data = json.loads(body)
    job_id = data["job_id"]

    print("Worker1 processing", job_id)

    time.sleep(3)

    result = f"processed-{job_id}"

    cur.execute(
        "UPDATE jobs SET result1=%s WHERE id=%s",
        (result, job_id)
    )
    conn.commit()

    message = json.dumps({
        "job_id": job_id,
        "result1": result
    })

    ch.basic_publish(
        exchange="",
        routing_key="worker2_queue",
        body=message,
        properties=pika.BasicProperties(delivery_mode=2),
    )

    ch.basic_ack(delivery_tag=method.delivery_tag)


connection = pika.BlockingConnection(pika.ConnectionParameters("rabbitmq"))
channel = connection.channel()
channel.queue_declare(queue="job_queue", durable=True)
channel.queue_declare(queue="worker2_queue", durable=True)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue="job_queue", on_message_callback=callback)

print("Worker1 started")
channel.start_consuming()

Notice something very important at the end:

ch.basic_ack(delivery_tag=method.delivery_tag)

RabbitMQ does not remove the message from the queue until the worker acknowledges it. So, if the worker crashes before sending the ACK, RabbitMQ simply requeues the message and another worker can pick it up. That’s how we avoid lost jobs.

Worker 2: The Second Stage

Worker2 consumes the result produced by Worker1. It finishes the job and marks it completed.

import json
import time
import pika
import psycopg2


def get_connection():
    return psycopg2.connect(
        host="postgres", database="jobsdb", user="user", password="password"
    )


def stage2_process(stage1_result):
    print("[Worker2] Processing stage 2")
    time.sleep(3)
    return f"FINAL({stage1_result})"


def callback(ch, method, properties, body):
    data = json.loads(body)
    job_id = data["job_id"]
    client_id = data["client_id"]

    try:
        conn = get_connection()
        cur = conn.cursor()
        cur.execute("SELECT stage1_result FROM jobs WHERE id=%s;", (job_id,))
        stage1_result = cur.fetchone()[0]

        final_result = stage2_process(stage1_result)

        cur.execute(
            "UPDATE jobs SET final_result=%s, status='completed' WHERE id=%s;",
            (final_result, job_id),
        )
        conn.commit()
        conn.close()

        ch.basic_publish(
            exchange="",
            routing_key="notification_queue",
            body=json.dumps({"job_id": job_id, "client_id": client_id}),
            properties=pika.BasicProperties(delivery_mode=2),
        )

        ch.basic_ack(delivery_tag=method.delivery_tag)

    except Exception as e:
        print("Error:", e)


connection = pika.BlockingConnection(pika.ConnectionParameters("rabbitmq"))
channel = connection.channel()
channel.queue_declare(queue="worker2_queue", durable=True)
channel.queue_declare(queue="notification_queue", durable=True)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue="worker2_queue", on_message_callback=callback)

print("Worker2 started")
channel.start_consuming()

Then Worker2 pushes a message to the notification queue.

notification_queue

Now the job pipeline is complete.

The Notification Service

Instead of polling the API constantly, the client connects to a WebSocket. When Worker2 finishes a job, the notification service sends a message to the client.

Job abc123 completed

The notification service listens to RabbitMQ:

def callback(ch, method, properties, body):
    data = json.loads(body)

    job_id = data["job_id"]

    asyncio.run(send_notification(job_id))

    ch.basic_ack(delivery_tag=method.delivery_tag)

And pushes the update to connected clients. This gives us real-time updates without constant polling.

Hiding Delay From Users In Systems With replica

In heavy read/write production systems, reads and writes are often separated.

Primary DB -> handles writes
Replica DB -> handles reads

Replicas are eventually consistent. This means there can be a small delay before the replica sees the new data. So what happens if we notify the user before the replica catches up? They might see:

Job completed

But when they request the result:

Result not found

That’s confusing and against good user experience. Instead, the client simply retries in background until it get the result.

function fetchResult(jobId) {
  fetch(`/jobs/${jobId}`)
    .then(r => r.json())
    .then(data => {
      if (!data.result) {
        setTimeout(() => fetchResult(jobId), 1000);
      } else {
        showResult(data.result);
      }
    });
}

The retry hides replica lag from the user. The user never notices the delay.

Our System is Resilient

There are several subtle decisions that make this architecture robust. Queues are durable. Messages are persistent. Workers send ACKs only after finishing work. If a worker crashes, RabbitMQ requeues the message automatically. Each service is isolated in its own container.

Finally docker Compose orchestrates everything:

version: "3.9"

services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
      POSTGRES_DB: jobsdb
    ports:
      - "5432:5432"

  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"
      - "15672:15672"

  api:
    build: ./api
    depends_on:
      - postgres
      - rabbitmq
    ports:
      - "8000:8000"

  worker1:
    build: ./worker1
    depends_on:
      - rabbitmq
      - postgres

  worker2:
    build: ./worker2
    depends_on:
      - rabbitmq
      - postgres

  notification:
    build: ./notification
    depends_on:
      - rabbitmq
    ports:
      - "8001:8001"

You can find the complete working project in my GitHub: async_jobs_pipeline_example
In complete version of project, in addition to final stage, each stage sends the status to client, so the user can see the real-time status of its task when each worker accomplished its job:

I hope you enjoyed this post :).