redis-message-queue 9.1.0

Python message queuing with Redis and message deduplication

redis-message-queue

Lightweight Python message queuing with Redis and built-in publish-side deduplication. Deduplicate publishes within a TTL window, with crash recovery (at-least-once) on by default — across any number of producers and consumers.

pip install "redis-message-queue>=9.1.0,<10.0.0"

Requires Python >= 3.12 and Redis server >= 6.2. Works with standalone Redis and Redis Sentinel; Redis Cluster is supported when the queue name is hash-tagged (e.g. {myqueue}) so all of a queue's keys share one slot — see Redis Cluster requirements and Sentinel setup.

redis-message-queue follows semantic versioning: breaking API changes land only in a new major version — hence the major-version upper bound in the install command above — so minor and patch upgrades are safe to take. See UPGRADING.md for per-major migration guides.

Mental model: redis-message-queue is a payload queue, not a task framework. Producers publish a str or dict; consumers decide what it means. There is no task registry, result backend, scheduler, or handler-level retry policy — and an ordinary exception raised inside a handler is terminal, not an automatic retry. Coming from Celery, RQ, Dramatiq, or taskiq? Read Migrating from task frameworks before porting code.

Quickstart

Redis must be running locally first: use redis-server or docker run -it --rm -p 6379:6379 redis:7.

Local Redis data: The sync and async quickstarts below connect to redis://localhost:6379/0 and use the fixed queue namespace quickstart. Each snippet publishes a message, then claims and removes one message under that namespace. If local DB 0 already contains quickstart data that matters, use a disposable Redis instance, a separate DB/port, or change the URL/queue name before running them.

import json
from uuid import uuid4
from redis import Redis
from redis_message_queue import RedisMessageQueue

client = Redis.from_url("redis://localhost:6379/0", decode_responses=True)
queue = RedisMessageQueue(
    "quickstart",
    client=client,
    deduplication=True,
    get_deduplication_key=lambda msg: msg["id"],
)
message = {"id": f"msg-{uuid4().hex}", "text": "hello"}
queue.publish(message)
with queue.process_message() as received:
    if received is not None:
        payload = json.loads(received)
        print(f"got {payload['text']}")
# Expected output: got hello

RedisMessageQueue itself is not a context manager. Use with queue.process_message() as received: for each message.

Sync handlers must be synchronous. If your handler is async def or returns any awaitable, use redis_message_queue.asyncio.RedisMessageQueue, or the sync process_message_callback(handler), which raises TypeError instead of dropping an unawaited coroutine while the message is acked. See Callback-style consuming.

Async quickstart

import asyncio
import json
from uuid import uuid4
from redis.asyncio import Redis
from redis_message_queue.asyncio import RedisMessageQueue

async def main():
    client = Redis.from_url("redis://localhost:6379/0", decode_responses=True)
    queue = RedisMessageQueue(
        "quickstart",
        client=client,
        deduplication=True,
        get_deduplication_key=lambda msg: msg["id"],
    )
    message = {"id": f"msg-{uuid4().hex}", "text": "hello"}
    await queue.publish(message)
    async with queue.process_message() as received:
        if received is not None:
            payload = json.loads(received)
            print(f"got {payload['text']}")
    await client.aclose()

asyncio.run(main())  # Expected output: got hello

Why redis-message-queue

The problem: You're sending messages between services or workers and need guarantees. Simple Redis LPUSH/BRPOP loses messages on crashes, doesn't deduplicate, and gives you no visibility into what succeeded or failed.

The solution: Atomic Lua scripts for publish + dedup, a processing queue for in-flight tracking (with optional crash recovery via visibility timeouts), and optional success/failure logs for observability.

Compared with rolling your own consumer groups on Redis Streams, you get publish-side deduplication, visibility-timeout redelivery, and a dead-letter queue out of the box on plain Redis lists and Lua — no Streams or consumer-group setup, and no separate broker like Kafka or SQS to run.

Feature	Details
Deduplicated publish	Lua-scripted atomic SET NX + LPUSH prevents duplicate enqueues within a configurable TTL window (default: 1 hour), even with producer retries. Requires an explicit get_deduplication_key callable so your application defines what counts as a duplicate. Note: deduplication is publish-side only and does not prevent duplicate delivery under at-least-once visibility-timeout reclaim
Visibility-timeout redelivery	Crashed or stalled consumers' messages are reclaimed and redelivered when a visibility timeout is configured
Completed & failed queues	Optional completed/failed queues for auditing, inspection, and application-owned manual reprocessing, with configurable max length to prevent unbounded growth
Dead-letter queue	Poison messages that exceed a configurable delivery count are automatically routed to a dead-letter queue instead of being redelivered indefinitely
Graceful shutdown	Built-in interrupt handler lets consumers finish current work before stopping
Lease heartbeats	Optional background lease renewal keeps long-running handlers from being redelivered prematurely
Connection retries	Exponential backoff with jitter for Redis ops; idempotent paths (deduplicated publish, ack, lease renewal, claim recovery) replay safely under retries, while non-deduplicated publish is intentionally not retried so the caller decides whether to retry (accepting potential duplicates). See Custom gateway
Async support	Mirrored async variant — same method and parameter names, with non-interchangeable callbacks. See Async API for the import swap and callback rules

All features are optional and can be enabled or disabled as needed.

Performance: throughput is essentially your Redis throughput — each publish, claim, and ack is a single atomic Lua round-trip to Redis with no separate broker process in the path (idle consumers add lightweight timed-wait polls). There are no published throughput benchmarks; size against your Redis instance's latency and ops/sec.

Delivery semantics

Configuration	Delivery guarantee
Default (visibility_timeout_seconds=300)	At-least-once — expired messages are reclaimed and redelivered
With visibility_timeout_seconds=None, max_delivery_count=None	At-most-once — a consumer crash loses the in-flight message

See Crash recovery with visibility timeout for details and tradeoffs. Because delivery-count limits depend on visibility-timeout reclaim, disabling lease-based crash recovery requires setting both visibility_timeout_seconds=None and max_delivery_count=None. Because at-least-once means the same payload can be delivered more than once, make your side effects idempotent — see Making consumers idempotent.

Both rows above describe consumer crashes. Durability across a Redis restart, failover, or maxmemory eviction is a separate concern, and is only as strong as your Redis persistence/replication configuration: a successful publish() can still be lost, because the library issues ordinary Redis writes and never calls WAIT or waits for an fsync or replica acknowledgement. See Known limitations in docs/operations.md.

Important: Ordinary Exception subclasses raised by handler code are terminal. This library is a payload queue, not a task framework: raising an ordinary Exception inside process_message() does not requeue the message. With enable_failed_queue=False, the message is removed from processing; with enable_failed_queue=True, it is moved to the failed list.

Fatal BaseException paths such as KeyboardInterrupt, SystemExit, and externally cancelled async tasks (asyncio.CancelledError) are shutdown/cancellation paths, not failed handler work. They can leave the message in processing for visibility-timeout reclaim, or orphan it when visibility_timeout_seconds=None, max_delivery_count=None; see Graceful shutdown and Abandoned in-flight messages.

Configuration

Every feature is optional and set through constructor arguments. The complete reference — with runnable snippets for each option — lives in docs/configuration.md:

• Deduplication — publish-side dedup keys, TTL windows, and cardinality guidance
• Success and failure tracking — optional completed/failed audit lists and their caps
• Publish backpressure — max_pending_length and the raise / drop_oldest / block overload policies
• Crash recovery with visibility timeout — leases, heartbeats, and redelivery
• Ordering and multi-consumer fairness — the claim-order guarantee and its limits
• Dead-letter queue — routing poison messages off the redelivery path
• Graceful shutdown — drain() and the three shutdown shapes
• Custom gateway — tuning retries and dedup TTL, or subclassing for new semantics
• Connection pool sizing — sizing Redis pools for heartbeat concurrency

For a single-page table of every constructor parameter, public method, and exported exception/type, see docs/api-reference.md.

Async API

Replace the import to use the async variant — it mirrors the sync API with the same method and parameter names (call the awaitable methods with await):

from redis_message_queue.asyncio import RedisMessageQueue

The sync and async classes intentionally share names. In modules that use both, alias the imports explicitly, for example from redis_message_queue import RedisMessageQueue as SyncRedisMessageQueue and from redis_message_queue.asyncio import RedisMessageQueue as AsyncRedisMessageQueue.

Callbacks are not interchangeable between the two classes: the sync queue rejects async callables, and on the async queue on_event must be async, while get_deduplication_key and on_heartbeat_failure may be sync or async.

The examples otherwise work the same way. One lifecycle rule to remember: the queue's drain() never closes the Redis client — calling client.close() / await client.aclose() stays your job (the async quickstart above shows it). See the API reference for the full drain contract.

Migrating from RQ / Celery / Dramatiq / taskiq

redis-message-queue is a payload queue, not a task framework. It has no task registry, job object, result backend, scheduler, workflow canvas, callback graph, or handler-level retry policy. Producers publish a str or dict payload, and consumers decide what that payload means.

The most important semantic differences from sibling task libraries are:

• Ordinary Exception subclasses raised by handler code are terminal. Raising an ordinary Exception inside process_message() removes the message from processing, or moves it to the failed list when enable_failed_queue=True; it does not requeue or retry the message. Fatal BaseException shutdown or cancellation paths are covered by Graceful shutdown and Abandoned in-flight messages.
• visibility_timeout_seconds is a crash/stall recovery lease, not a runtime limit. Slow handlers are not interrupted; after the lease expires another consumer can process the same payload concurrently.
• on_event is telemetry only. Callback exceptions are logged and emitted as RuntimeWarning, but they do not affect ack/nack, failed-queue movement, or any other message outcome. Do not use on_event for sagas, follow-up writes, billing callbacks, or other correctness-critical work.
• Dict payloads are JSON data, not Python call arguments. JSON does not preserve every Python type: tuples become lists, and sets or custom objects raise unless you encode them into JSON-native values first.
• Process-global signal ownership cannot be safely chained with Celery, RQ, or Dramatiq CLI workers. Prefer one top-level owner that calls queue.drain() or sets an application stop event, and run sibling workers in separate processes.

When migrating on the same Redis deployment, prefer separate Redis DBs or hard namespaces. Do not point a Celery, RQ, Dramatiq, or taskiq worker at an rmq pending key. A sibling worker can pop the rmq stored message, fail its own decoder, and leave the rmq queue without that message. Also avoid custom key_separator values that synthesize another library's key namespace, such as using ":queue:" with a queue name that overlaps RQ keys. rmq has no fixed library prefix; generated keys share the Redis DB namespace with every other Redis user.

Set strict_envelope_decoding=True if this Redis is shared with sibling task libraries (Celery, RQ, Dramatiq) to fail-fast on foreign payloads. With the default False, non-rmq values that do not start with the rmq envelope prefix remain backward-compatible raw messages and are yielded to the handler.

Production notes

Deploying to production? See docs/operations.md for fork safety and pre-fork servers (gunicorn --preload, multiprocessing, ProcessPoolExecutor) and Redis memory sizing for deduplication and replay metadata. To inspect or manage live queues without reaching for raw Redis commands — list depths, peeking without consuming, redriving the dead-letter queue, and purging — see Inspecting and managing queues and the Redis key layout reference.

Production patterns

Runnable, production-shaped examples (each has a sync version and an asyncio/ sibling):

• backpressure.py (async) — publish under a bounded pending queue, retrying on QueueBackpressureError instead of growing memory unbounded.
• graceful_shutdown.py (async) — finish the in-flight message and stop cleanly on SIGINT/SIGTERM.
• observability.py (async) — wire the on_event callback to metrics/logs with secret-safe event.error handling.
• idempotent_consumer.py (async) — guard side effects with SET NX EX so at-least-once redelivery is safe.

Observability

redis-message-queue emits lifecycle events through an optional on_event callback — publish, dedup hits, claims, reclaims, ack/nack, DLQ moves, heartbeats, drain, and retries — and exposes a typed exception hierarchy rooted at RedisMessageQueueError. The full guide (event catalog, dispatch context, timing versus Redis commit, intentionally silent paths, secret-safety for event.error, and the exception tree) is in docs/observability.md.

Known limitations

Known limitations and edge cases — timed-wait polling, Lua atomicity, batch-reclaim bounds, Redis Cluster hash-tag requirements, non-ASCII payload sizing, and client-side Retry interactions — are catalogued in docs/operations.md#known-limitations. For the full residual-risk register, see docs/production-readiness.md.

Troubleshooting

Seeing RetryBudgetExhaustedError, WRONGTYPE, stuck/duplicate deliveries, a filling DLQ, or CROSSSLOT errors? The symptom-keyed index in docs/troubleshooting.md points to the relevant deep-dive section for each.

Upgrading

Version migration guides — v7→v8, v6→v7, v5→v6, v2→v3, and the destructive-on-live-queues configuration changes — are in UPGRADING.md. Per-release detail lives in CHANGELOG.md.

Running locally

These examples ship in the GitHub repo, not the PyPI package — clone the repo and run uv sync first.

Start a local Redis server with redis-server, or with Docker:

docker run -it --rm -p 6379:6379 redis:7

Try the examples with multiple terminals:

These examples connect to REDIS_URL when it is set; otherwise they use redis://localhost:6379/0 (database 0). The send and receive examples use the fixed queue name my_message_queue: publishers write queue data under that namespace, including pending and deduplication keys, and consumers can claim and remove messages from it. If existing local Redis data in localhost:6379/0 matters, run a disposable Redis instance or select a separate Redis database before running the commands.

# Two publishers
uv run python -m examples.send_messages
uv run python -m examples.send_messages

# Three consumers
uv run python -m examples.receive_messages
uv run python -m examples.receive_messages
uv run python -m examples.receive_messages

These publisher and consumer examples are long-running; stop them with Ctrl+C or another interrupt when you are done. Publishers print Success: Sent message ... or Duplicate: Message .... Consumers print Received Message: ... before simulated time.sleep(...) work and Finished processing message ... afterward. On a clean single handled shutdown, the consumer prints Exiting... and the signal handler prints Received signal: ... on stderr.

When examples run through wrappers such as uv run, terminal interrupts may reach the process group more than once. If an interrupt lands during the consumer's simulated time.sleep(...) work, you can see a KeyboardInterrupt traceback instead of the clean Exiting... line.

Contributing

Clone the repo, then:

uv sync                       # install dependencies
uv run pytest                 # run the test suite
uv run ruff check .           # lint
uv run ruff format --check .  # check formatting

Unit tests run against an in-memory fake Redis, so uv run pytest needs no external services. Tests marked integration connect to a real Redis at $REDIS_URL (default redis://localhost:6379/15) and skip automatically when none is reachable.

License

Released under the MIT License.