Phase 2: Flask + Redis Async Job Queue System

🏋️ Objective

Expand your Flask + Redis setup into a real async processing system by building a Redis-backed worker, adding a status-check endpoint, and simulating real-world job processing.

🔄 Architecture Overview

Client ➔ Flask API (enqueue) ➔ Redis Queue ➔ Python Worker ➔ Redis Job Store ➔ Flask API (job status)

• Clients POST jobs to /enqueue
  
  

• Jobs get added to Redis list jobs
  
  

• worker.py continuously polls Redis, processes jobs
  
  

• Updates job status in Redis (by job_id)
  
  

• Clients can GET /job/<id> to check progress
  
  

✅ Step 1: Add Job Status API Endpoint

File: app.py (add below your /enqueue route)

@app.route("/job/<job_id>", methods=["GET"])

def get_job_status(job_id):

    job_key = f"job:{job_id}"

    job_data = r.get(job_key)

    if not job_data:

        return jsonify({"error": "Job not found"}), 404

    return jsonify(json.loads(job_data))

This lets clients poll the job status by ID.

✅ Step 2: Build Your Redis Worker

File: worker.py

import redis

import time

import json

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

print("Worker is running...")

while True:

    job_data = r.lpop("jobs")

    if job_data:

        job = json.loads(job_data)

        job_id = job["id"]

        print(f"Processing job: {job_id}")

        # Simulate status update to 'processing'

        job["status"] = "processing"

        r.set(f"job:{job_id}", json.dumps(job))

        # Simulate processing work

        time.sleep(5)  # fake long-running task

        # Update to 'done'

        job["status"] = "done"

        r.set(f"job:{job_id}", json.dumps(job))

        print(f"Job {job_id} complete")

    else:

        time.sleep(1)  # no jobs, wait a bit

📝 Step 3: Test Workflow

✉️ 1. Start Everything

Terminal 1 (Flask API):

$ python app.py

Terminal 2 (Worker):

$ python worker.py

Terminal 3:

$ curl -X POST http://localhost:5000/enqueue \

  -H "Content-Type: application/json" \

  -d '{"task": "process_doc", "doc_id": 42}'

# Response:

{ "job_id": "abcd-1234", "status": "queued" }

📊 2. Check Status

$ curl http://localhost:5000/job/abcd-1234

# Initially:

{ "id": "abcd-1234", "status": "queued" }

# After a few seconds:

{ "id": "abcd-1234", "status": "done" }

📆 Bonus: requirements.txt

flask

redis

🌟 Phase 2 Outcome

You now have a:

• Real Flask API that queues and monitors jobs
  
  

• Redis-backed system for decoupling work
  
  

• Standalone Python worker that processes tasks
  
  

🎓 Phase 3 Preview: Scaling + Auth

• Add user tracking (auth/session per job)
  
  

• Add retries / error state
  
  

• Queue priority handling (push left or right)
  
  

• Async workers (Celery or custom ThreadPool)
  
  

Say "Ready for Phase 3" when you want to push further.