# Redis job queue with redis-py ```json metadata { "title": "Redis job queue with redis-py", "description": "Implement a Redis job queue in Python with redis-py", "categories": ["docs","develop","stack","oss","rs","rc"], "tableOfContents": {"sections":[{"id":"overview","title":"Overview"},{"id":"how-it-works","title":"How it works"},{"children":[{"id":"data-model","title":"Data model"}],"id":"the-job-queue-helper","title":"The job queue helper"},{"id":"enqueueing-jobs","title":"Enqueueing jobs"},{"id":"claiming-jobs-with-brpoplpush","title":"Claiming jobs with BRPOPLPUSH"},{"id":"completing-jobs","title":"Completing jobs"},{"id":"failing-and-retrying","title":"Failing and retrying"},{"id":"reclaiming-stuck-jobs","title":"Reclaiming stuck jobs"},{"id":"stats-and-history","title":"Stats and history"},{"id":"prerequisites","title":"Prerequisites"},{"children":[{"id":"get-the-source-files","title":"Get the source files"},{"id":"start-the-demo-server","title":"Start the demo server"}],"id":"running-the-demo","title":"Running the demo"},{"id":"the-mock-worker-pool","title":"The mock worker pool"},{"children":[{"id":"choose-a-visibility-timeout-that-matches-your-worst-case-job-latency","title":"Choose a visibility timeout that matches your worst-case job latency"},{"id":"run-the-reclaimer-on-a-schedule","title":"Run the reclaimer on a schedule"},{"id":"use-a-separate-redis-database-or-key-prefix-per-queue","title":"Use a separate Redis database or key prefix per queue"},{"id":"cap-the-completed-and-failed-history","title":"Cap the completed and failed history"},{"id":"tune-maxattempts-per-job-kind","title":"Tune max_attempts per job kind"},{"id":"inspect-queue-state-directly-in-redis","title":"Inspect queue state directly in Redis"}],"id":"production-usage","title":"Production usage"},{"id":"learn-more","title":"Learn more"}]} , "codeExamples": [] } ``` This guide shows you how to implement a Redis-backed job queue in Python with [`redis-py`](https://redis.io/docs/latest/develop/clients/redis-py). It includes a small local web server built with the Python standard library so you can enqueue jobs, watch a pool of workers drain them, and see the reclaimer recover jobs from a simulated worker crash. ## Overview A job queue lets your application offload background work — sending email, processing payments, image transcoding, ML inference, webhooks — from the request path. Producers enqueue jobs in milliseconds and return to the user; workers pull from the queue and process them on their own schedule. That gives you: * Low-latency user-facing requests, even when downstream work is slow or bursty * Horizontal scale across many worker processes that share one Redis instance * At-least-once delivery so a worker crash doesn't lose work * Visibility-timeout reclaim that returns stuck jobs to the queue automatically * Job metadata, retry counts, and completion results in Redis hashes with TTL In this example, each job is identified by a random hex ID and its payload, status, and result live in a Redis hash under `queue:jobs:job:{id}`. Pending IDs sit in a list, claimed IDs move atomically to a *processing* list, and completed or failed IDs land in short history lists. ## How it works The flow looks like this: 1. The application calls `queue.enqueue(payload)` 2. The helper writes the job metadata hash and `LPUSH`es the job ID onto the pending list 3. A worker calls `queue.claim(timeout_ms)` 4. The helper runs `BRPOPLPUSH` to atomically move the next pending ID into the processing list and writes a per-claim `claim_token` plus `claimed_at_ms` on the hash 5. The worker runs the job and calls `queue.complete(job, result)` or `queue.fail(job, error)` 6. `complete` removes the job from the processing list, writes the result, and `LPUSH`es the ID onto the completed history (with `LTRIM` and an `EXPIRE` on the hash for cleanup) 7. `fail` either retries the job (back to pending) or moves it to the failed list once retries are exhausted If a worker dies before completing a job, the job sits in the processing list with a `claimed_at_ms` older than the visibility timeout. A periodic call to `queue.reclaim_stuck()` finds those jobs and moves them back to pending so another worker can pick them up. Every state change holds the token: a worker that has been reclaimed cannot later complete or fail a job another worker has already claimed. ## The job queue helper The `RedisJobQueue` class wraps the queue operations ([source](https://github.com/redis/docs/blob/main/content/develop/use-cases/job-queue/redis-py/job_queue.py)): ```python import redis from job_queue import RedisJobQueue r = redis.Redis(host="localhost", port=6379, decode_responses=True) queue = RedisJobQueue(redis_client=r, visibility_ms=5000) job_id = queue.enqueue({"kind": "email", "recipient": "alice@example.com"}) # In a worker process: job = queue.claim(timeout_ms=1000) if job is not None: try: # ... run the job ... queue.complete(job, result={"sent_at": "2026-05-11T15:00:00Z"}) except Exception as exc: queue.fail(job, error=str(exc)) # In a periodic sweeper: reclaimed = queue.reclaim_stuck() ``` ### Data model Each job's state lives in a Redis hash plus a position in one of four lists: ```text queue:jobs:pending (list) pending job IDs, oldest at the right queue:jobs:processing (list) claimed but not yet completed queue:jobs:completed (list) recent successes (LTRIM-capped history) queue:jobs:failed (list) terminally failed jobs queue:jobs:job:{id} (hash) per-job metadata queue:jobs:events (pubsub) completion notifications ``` A job's hash carries: ```text queue:jobs:job:9a4f... id = 9a4f... payload = {"kind":"email","recipient":"alice@example.com"} status = pending | processing | completed | failed attempts = 1 enqueued_at_ms = 1715441000000 claimed_at_ms = 1715441000123 claim_token = b3c0d1e2... (per-claim random token) completed_at_ms = 1715441000456 result = {"sent_at":"..."} last_error = "smtp timeout" ``` The implementation uses: * [`LPUSH`](https://redis.io/docs/latest/commands/lpush) to add new job IDs to the pending list * [`BRPOPLPUSH`](https://redis.io/docs/latest/commands/brpoplpush) to atomically claim a job into the processing list * [`LREM`](https://redis.io/docs/latest/commands/lrem) to remove a claimed job from the processing list on complete or fail * [`LTRIM`](https://redis.io/docs/latest/commands/ltrim) to cap the completed and failed history lists * [`HSET`](https://redis.io/docs/latest/commands/hset) / [`HGETALL`](https://redis.io/docs/latest/commands/hgetall) for job metadata * [`EXPIRE`](https://redis.io/docs/latest/commands/expire) on completed and failed hashes for automatic cleanup * [`PUBLISH`](https://redis.io/docs/latest/commands/publish) on `queue:jobs:events` for completion signalling * [Lua scripting](https://redis.io/docs/latest/develop/programmability/eval-intro) ([`EVALSHA`](https://redis.io/docs/latest/commands/evalsha)) for the complete, fail, and reclaim flows so each runs atomically against the processing list and metadata hash ## Enqueueing jobs `enqueue()` writes the metadata hash and pushes the job ID onto the pending list in one pipeline: ```python def enqueue(self, payload: dict) -> str: job_id = secrets.token_hex(8) now_ms = self._now_ms() meta = { "id": job_id, "payload": json.dumps(payload), "status": "pending", "attempts": 0, "enqueued_at_ms": now_ms, "claim_token": "", } pipe = self.redis.pipeline() pipe.hset(self._meta_key(job_id), mapping=meta) pipe.lpush(self.pending_key, job_id) pipe.execute() return job_id ``` The payload is stored as JSON so the queue can carry arbitrary nested structures without forcing every field into a hash. ## Claiming jobs with BRPOPLPUSH A worker blocks until a job is available, then atomically pops it from the pending list and pushes it onto the processing list. `BRPOPLPUSH` does both in a single Redis call: ```python def claim(self, timeout_ms: int = 1000) -> Optional[ClaimedJob]: timeout_s = max(timeout_ms / 1000.0, 0.1) job_id = self.redis.brpoplpush(self.pending_key, self.processing_key, timeout=timeout_s) if job_id is None: return None token = secrets.token_hex(8) now_ms = self._now_ms() meta_key = self._meta_key(job_id) pipe = self.redis.pipeline() pipe.hset(meta_key, mapping={ "status": "processing", "claimed_at_ms": now_ms, "claim_token": token, }) pipe.hincrby(meta_key, "attempts", 1) pipe.hgetall(meta_key) _, _, meta = pipe.execute() return ClaimedJob(job_id, json.loads(meta["payload"]), int(meta["attempts"]), token) ``` The `claim_token` is the worker's proof of ownership for this attempt. Every subsequent state change (complete, fail) checks it before touching the processing list, so a worker that hung past the visibility timeout cannot interfere with the new claimant. ## Completing jobs `complete()` runs a Lua script so the processing-list removal, the metadata write, and the history push happen atomically: ```python def complete(self, job: ClaimedJob, result: dict) -> bool: ok = self._complete( keys=[self.meta_prefix, self.processing_key, self.completed_key], args=[ job.id, job.claim_token, "completed", self._now_ms(), json.dumps(result), self.completed_ttl, self.completed_history, ], ) if not ok: return False self.redis.publish(self.events_channel, json.dumps({"id": job.id, "status": "completed"})) return True ``` The Lua script checks the token first and returns `0` if the worker no longer owns the job (because the reclaimer moved it back to pending). The metadata hash also gets an `EXPIRE` so completed jobs are cleaned up automatically. ## Failing and retrying `fail()` either retries the job (back to pending) or moves it to the failed list once retries are exhausted: ```python def fail(self, job: ClaimedJob, error: str) -> bool: retry = job.attempts < self.max_attempts result = self._fail( keys=[self.meta_prefix, self.processing_key, self.pending_key, self.failed_key], args=[ job.id, job.claim_token, error, self._now_ms(), self.completed_ttl, self.completed_history, "1" if retry else "0", ], ) return bool(result) ``` The attempt counter is incremented on every `claim()`, so a job that fails three times is moved to the failed list with `attempts = 3` and the final `last_error` preserved. ## Reclaiming stuck jobs If a worker dies mid-job — the process is killed, the host loses power, the network partitions — the job sits in the processing list with a `claimed_at_ms` that never advances. A periodic call to `reclaim_stuck()` walks the processing list and moves any job past the visibility timeout back to pending: ```python def reclaim_stuck(self) -> list[str]: reclaimed = self._reclaim( keys=[self.pending_key, self.processing_key, self.meta_prefix], args=[self._now_ms(), self.visibility_ms], ) return list(reclaimed) ``` The Lua script also handles a narrower race: a worker that crashed between `BRPOPLPUSH` and writing `claimed_at_ms`. Those jobs are reclaimed after `2 × visibility_ms` using `enqueued_at_ms` as a fallback timer, so they aren't stranded forever. ## Stats and history `stats()` reports queue depth plus per-process counters: ```python def stats(self) -> dict: pipe = self.redis.pipeline() pipe.llen(self.pending_key) pipe.llen(self.processing_key) pipe.llen(self.completed_key) pipe.llen(self.failed_key) pending, processing, completed, failed = pipe.execute() return { "enqueued_total": self._enqueued, "completed_total": self._completed, "failed_total": self._failed, "reclaimed_total": self._reclaimed, "pending_depth": pending, "processing_depth": processing, "completed_depth": completed, "failed_depth": failed, "visibility_ms": self.visibility_ms, } ``` The completed and failed lists are capped via `LTRIM` so they never grow unbounded; a real deployment would also write completion events to a longer-term audit log if needed. ## Prerequisites * Redis 6.2 or later running locally on the default port (6379). Earlier versions still work, since the helper uses commands that have existed since Redis 2.6. * Python 3.9 or later. * The `redis-py` client. Install it with: ```bash pip install "redis>=5.0" ``` ## Running the demo ### Get the source files The demo consists of three Python files. Download them from the [`redis-py` source folder](https://github.com/redis/docs/tree/main/content/develop/use-cases/job-queue/redis-py) on GitHub, or grab them with `curl`: ```bash mkdir job-queue-demo && cd job-queue-demo BASE=https://raw.githubusercontent.com/redis/docs/main/content/develop/use-cases/job-queue/redis-py curl -O $BASE/job_queue.py curl -O $BASE/worker.py curl -O $BASE/demo_server.py ``` ### Start the demo server From that directory: ```bash python3 demo_server.py ``` You should see: ```text Redis job-queue demo server listening on http://127.0.0.1:8090 Using Redis at localhost:6379 Visibility timeout: 5000 ms ``` Open [http://127.0.0.1:8090](http://127.0.0.1:8090) in a browser. You can: * Enqueue jobs of different kinds (email, webhook, thumbnail, invoice) in batches. * Start a pool of workers with configurable size, work latency, and *failure* / *hang* rates. A non-zero hang rate simulates worker crashes. * Click **Run reclaim sweep** to move any timed-out processing jobs back to pending. * Watch pending / processing / completed / failed lists update every 800 ms. If your Redis server is running elsewhere, start the demo with `--redis-host` and `--redis-port`. You can also tune the visibility timeout with `--visibility-ms`. ## The mock worker pool The demo includes a small `Worker` and `WorkerPool` ([source](https://github.com/redis/docs/blob/main/content/develop/use-cases/job-queue/redis-py/worker.py)) that stands in for whatever real background work your application would run. Each worker: * Blocks on `queue.claim()` for new jobs. * Sleeps `work_latency_ms` to simulate doing the work. * Either completes successfully, fails (calling `queue.fail()`), or *hangs* — returning without completing or failing the job so the reclaimer has to recover it. The `fail_rate` and `hang_rate` knobs let you watch the at-least-once delivery and reclaim behaviours from the UI without writing test code. ## Production usage ### Choose a visibility timeout that matches your worst-case job latency The visibility timeout has to exceed the longest real job time, with margin. If it's too short, a healthy worker that's running a slow job will get its work duplicated when the reclaimer fires. If it's too long, a real crash takes longer to detect. Most production deployments use a per-queue value tuned to the 99th-percentile job latency — for example, 2 minutes for email and 30 minutes for video transcoding. ### Run the reclaimer on a schedule The demo only reclaims when you click the button. In production, run `reclaim_stuck()` from a periodic task (every few seconds for fast queues, every minute for slow ones), or from each worker before it blocks on `claim()`. Both patterns work as long as *someone* runs the sweep. ### Use a separate Redis database or key prefix per queue The helper takes a `queue_name` argument so you can run multiple independent queues against one Redis instance — for example, one queue per priority level, or one per job kind. Keep queue keys under a clearly-namespaced prefix (here, `queue:jobs:*`) so they're easy to inspect and easy to clear without touching application data. ### Cap the completed and failed history The demo keeps the last 50 completed and 50 failed job IDs via `LTRIM`. If you need longer history for audit purposes, write completion events to a separate Redis Stream (or to an external store) and keep the in-queue history short. Stream consumer groups give you the same fan-out semantics with a much richer history. ### Tune `max_attempts` per job kind A blanket `max_attempts = 3` is a reasonable default for transient failures (network timeouts, rate limits). Jobs that talk to non-idempotent external systems — for example, posting a Stripe charge — need either application-level idempotency keys or a much lower retry count. The helper exposes `max_attempts` so each queue can pick its own policy. ### Inspect queue state directly in Redis Because the queue is just lists and hashes, you can inspect it with `redis-cli`: ```bash # How many pending jobs? redis-cli LLEN queue:jobs:pending # Look at the next 5 jobs to be picked up. redis-cli LRANGE queue:jobs:pending -5 -1 # Read a job's metadata. redis-cli HGETALL queue:jobs:job:9a4f0d1c # How many jobs are currently being processed? redis-cli LLEN queue:jobs:processing # Clear everything for this queue (be careful — this deletes work). redis-cli --scan --pattern 'queue:jobs:*' | xargs redis-cli DEL ``` ## Learn more This example uses the following Redis commands: * [`LPUSH`](https://redis.io/docs/latest/commands/lpush) to enqueue a job ID. * [`BRPOPLPUSH`](https://redis.io/docs/latest/commands/brpoplpush) to atomically claim a job into the processing list. * [`LREM`](https://redis.io/docs/latest/commands/lrem) to remove a job from the processing list on complete or fail. * [`LRANGE`](https://redis.io/docs/latest/commands/lrange) and [`LLEN`](https://redis.io/docs/latest/commands/llen) to read queue depth and list contents. * [`LTRIM`](https://redis.io/docs/latest/commands/ltrim) to cap the completed and failed history. * [`HSET`](https://redis.io/docs/latest/commands/hset) and [`HGETALL`](https://redis.io/docs/latest/commands/hgetall) for job metadata. * [`HINCRBY`](https://redis.io/docs/latest/commands/hincrby) for the attempt counter. * [`EXPIRE`](https://redis.io/docs/latest/commands/expire) for automatic cleanup of completed and failed jobs. * [`PUBLISH`](https://redis.io/docs/latest/commands/publish) for job-completion notifications. * [`EVALSHA`](https://redis.io/docs/latest/commands/evalsha) for atomic complete, fail, and reclaim flows. See the [`redis-py` documentation](https://redis.io/docs/latest/develop/clients/redis-py) for full client reference.