Flask by example implementing a redis task queue

A task queue separates HTTP admission from background execution. Flask validates a small request and returns a job ID; a separate RQ worker reads the job from Redis, executes it, and stores its state. This tutorial implements that full lifecycle, including duplicates, a transient failure, and a permanently failed job.

The example is a local learning service, with deliberately exposed failure modes for testing. It is not ready to expose publicly. Tests used Python 3.12.14, Flask 3.1.3, RQ 2.12.0, redis-py 8.1.0, and an isolated Redis 6.2.14 process on macOS. The worker ran as a separate process; no application Redis or remote service was used. Redis 6.2.14 identifies the test fixture, not a recommended production release.

Prepare one folder and a disposable Redis

Use a dedicated virtual environment and install these Python packages:

python -m pip install Flask==3.1.3 rq==2.12.0 redis==8.1.0 requests==2.34.2

Start a disposable Redis instance bound to loopback, with no persistence, on an unused port. The following POSIX command uses port 6388; it fails if that port is unavailable. Do not point this tutorial at an existing shared Redis instance.

redis-server --bind 127.0.0.1 --port 6388 --save "" --appendonly no

In the web and worker terminals, set REDIS_URL to redis://127.0.0.1:6388/0. The code requires the setting rather than silently defaulting to a possibly shared service. Stop this disposable Redis process when finished. Production Redis needs an independently selected supported version, access controls, persistence and recovery decisions.

Put the importable task in tasks.py

The failure mode is an explicit teaching fixture. transient fails on the first attempt; permanent fails on every attempt. The successful calculation has no external side effects.

from rq import get_current_job

def summarize(numbers, mode="normal"):
    job = get_current_job()
    if job is None:
        raise RuntimeError("run this task through an RQ worker")
    attempts = job.meta.get("attempts", 0) + 1
    job.meta["attempts"] = attempts
    job.save_meta()
    if mode == "permanent" or (mode == "transient" and attempts == 1):
        raise RuntimeError("intentional fixture failure")
    return {"count": len(numbers), "total": sum(numbers), "attempts": attempts}

RQ imports the function by module path in the worker process, so keep it in an importable file rather than an interactive shell or a web request closure. Attempt metadata makes retries observable. It is not a transactional external-side-effect ledger.

Create app.py with admission and status endpoints

import os
from uuid import UUID
from flask import Flask, jsonify, request
from redis import Redis
from redis.exceptions import RedisError
from rq import Queue, Retry
from rq.exceptions import DuplicateJobError
from tasks import summarize

app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 8192
redis = Redis.from_url(os.environ["REDIS_URL"],
                       socket_connect_timeout=3, socket_timeout=3)
queue = Queue("tutorial", connection=redis, serializer="json")

@app.post("/jobs")
def submit():
    body = request.get_json()
    if not isinstance(body, dict) or set(body) - {"request_id", "numbers", "mode"}:
        return jsonify(error="invalid request object"), 400
    try:
        request_id = UUID(body["request_id"])
    except (KeyError, ValueError, TypeError, AttributeError):
        return jsonify(error="request_id must be a UUID string"), 400
    numbers, mode = body.get("numbers"), body.get("mode", "normal")
    if (not isinstance(numbers, list) or not 1 <= len(numbers) <= 100 or
        any(type(n) is not int or not -1000 <= n <= 1000 for n in numbers) or
        mode not in ("normal", "transient", "permanent")):
        return jsonify(error="invalid numbers or mode"), 400
    job_id = "tutorial-" + request_id.hex
    try:
        queue.enqueue(summarize, numbers, mode, job_id=job_id, unique=True,
                      job_timeout=10, result_ttl=300, failure_ttl=300,
                      retry=Retry(max=2))
    except DuplicateJobError:
        return jsonify(error="request_id already exists"), 409
    except RedisError:
        return jsonify(error="queue unavailable; admission outcome may be unknown"), 503
    return jsonify(job_id=job_id, status_url="/jobs/" + str(request_id)), 202

@app.get("/jobs/<uuid:request_id>")
def status(request_id):
    try:
        job = queue.fetch_job("tutorial-" + request_id.hex)
        if job is None:
            return jsonify(error="unknown or expired job"), 404
        job_state = job.get_status(refresh=True)
        if job_state is None:
            return jsonify(error="unknown or expired job"), 404
        state = job_state.value
        response = {"state": state, "attempts": job.meta.get("attempts", 0)}
        if state == "finished":
            response["result"] = job.return_value()
        elif state == "failed":
            response["error"] = "job failed; inspect trusted worker logs"
        return jsonify(response)
    except RedisError:
        return jsonify(error="queue unavailable"), 503

POST creates work; GET only reads state. Payload size, list length, integer range and allowed modes are bounded. A client-supplied request ID plus unique=True prevents another admission with the same ID while the job exists, returning 409 even if the repeated body is identical. The protection expires when the job expires or is deleted; it is not permanent exactly-once execution. After an uncertain 503, check status and retain the same request ID instead of immediately inventing a new one.

Both producer and worker use JSON serialization. This avoids RQ's default pickle deserialization, but Redis must still be trusted: job records contain callable paths and execution metadata. JSON is not authorization. This service lacks real user authentication, per-user job ownership, rate limits and capacity admission; add those before any public deployment and remove the artificial failure modes.

Run web and worker as separate processes

From the folder containing app.py and tasks.py, run these commands in separate POSIX terminals with the same environment. The web server binds to loopback and does not enable debug mode.

export REDIS_URL=redis://127.0.0.1:6388/0
python -m flask --app app run --host 127.0.0.1 --port 5001

export REDIS_URL=redis://127.0.0.1:6388/0
rq worker tutorial --url "$REDIS_URL" --serializer json

Without the worker, accepted jobs remain queued. A worker normally processes one job at a time. Use a proper WSGI server and worker process supervisor for deployment; Flask's development server is for this local demonstration. The CLI and worker-class behavior varies by platform, so test the chosen worker on the target OS.

Submit and poll a job from a third terminal

Run this client after both servers are ready. It stops after 30 polls and raises on an unsuccessful terminal state. The socket timeouts and finite poll count are separate limits; they are not a strict whole-job cancellation mechanism.

import os
import time
from uuid import uuid4
import requests

base = os.environ.get("DEMO_HTTP_URL", "http://127.0.0.1:5001")
payload = {"request_id": str(uuid4()), "numbers": [2, 3], "mode": "normal"}
with requests.Session() as client:
    client.trust_env = False
    response = client.post(base + "/jobs", json=payload, timeout=(2, 2))
    response.raise_for_status()
    print(response.status_code)
    status_url = base + response.json()["status_url"]
    for _ in range(30):
        status_response = client.get(status_url, timeout=(2, 2))
        status_response.raise_for_status()
        status = status_response.json()
        if status["state"] == "finished":
            print(status["state"], status["result"]["total"])
            break
        if status["state"] in {"failed", "stopped", "canceled"}:
            raise RuntimeError("job reached an unsuccessful terminal state")
        time.sleep(0.1)
    else:
        raise TimeoutError("job did not finish within 30 polls")
    duplicate = client.post(base + "/jobs", json=payload, timeout=(2, 2))
    print(duplicate.status_code)

Expected output:

202
finished 5
409

Observe normal, retry, duplicate, and failure outcomes

Submit a JSON object with a new UUID request_id, numbers [2,3], and mode normal to POST /jobs. Expect HTTP 202 and a status URL. Poll that URL with a bounded client interval and deadline. Terminal outcomes for the test fixture are:

Submitting the same UUID again returns 409. A list containing true, a number outside the allowed range, or an unknown mode returns 400. An unknown or expired UUID returns 404. Polling may skip brief intermediate states; do not require every run to observe started. The isolated test starts the worker after admission to prove the initial queued state.

Make retry and retention policies explicit

Retry(max=2) means two retries after the initial attempt: at most three attempts here, with immediate requeue. Production retries need failure classification and backoff; a permanent data error should not be retried blindly. Nonzero retry intervals require the appropriate RQ scheduler support. External writes need their own idempotency keys or transactional design because a worker can fail after a side effect and before recording success.

Results and failed-job records are retained for 300 seconds in this fixture. A 404 therefore means unknown or expired, not proof the job never ran. Worker tracebacks remain in trusted logs and Redis; the status response does not expose them. Queue backlog, job age, failure rate, retention and Redis memory need operational monitoring independently of this functional test.

References: RQ queues and unique jobs, workers and serializers, retry behavior, and Flask deployment guidance.