Metadata-Version: 2.4
Name: redis-message-queue
Version: 9.0.0
Summary: Python message queuing with Redis and message deduplication
Project-URL: Homepage, https://github.com/Elijas/redis-message-queue
Project-URL: Repository, https://github.com/Elijas/redis-message-queue
Project-URL: Issues, https://github.com/Elijas/redis-message-queue/issues
Author-email: Elijas <4084885+Elijas@users.noreply.github.com>
License: MIT
License-File: LICENSE
Keywords: deduplication,message-queue,redis,task-queue
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: System :: Distributed Computing
Requires-Python: <4.0,>=3.12
Requires-Dist: redis<9.0.0,>=5.0.1
Requires-Dist: tenacity<10,>=8.1.0
Description-Content-Type: text/markdown

# redis-message-queue

[![PyPI Version](https://img.shields.io/pypi/v/redis-message-queue?color=43cd0f&style=flat&label=pypi)](https://pypi.org/project/redis-message-queue)
[![PyPI Downloads](https://img.shields.io/pypi/dm/redis-message-queue?color=43cd0f&style=flat&label=downloads)](https://pypistats.org/packages/redis-message-queue)
[![License: MIT](https://img.shields.io/badge/License-MIT-43cd0f.svg?style=flat&label=license)](https://github.com/Elijas/redis-message-queue/blob/main/LICENSE)
[![Maintained: yes](https://img.shields.io/badge/yes-43cd0f.svg?style=flat&label=maintained)](https://github.com/Elijas/redis-message-queue/issues)
[![CI](https://github.com/Elijas/redis-message-queue/actions/workflows/ci.yml/badge.svg)](https://github.com/Elijas/redis-message-queue/actions/workflows/ci.yml)
[![codecov](https://codecov.io/gh/Elijas/redis-message-queue/graph/badge.svg)](https://codecov.io/gh/Elijas/redis-message-queue)
[![Linter: Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)

**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.

```bash
pip install "redis-message-queue>=9.0.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](https://github.com/Elijas/redis-message-queue/blob/main/docs/operations.md#known-limitations)
and [Sentinel setup](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#custom-gateway).

redis-message-queue follows [semantic versioning](https://semver.org): 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](https://github.com/Elijas/redis-message-queue/blob/main/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](#migrating-from-rq--celery--dramatiq--taskiq) 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.

```python
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 message:
    if message is not None:
        payload = json.loads(message)
        print(f"got {payload['text']}")
# Expected output: got hello
```

`RedisMessageQueue` itself is not a context manager. Use
`with queue.process_message() as message:` 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](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#callback-style-consuming).

### Async quickstart

```python
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 message:
        if message is not None:
            payload = json.loads(message)
            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](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#custom-gateway) |
| **Async support** | Mirrored async variant — same method and parameter names, with non-interchangeable callbacks. See [Async API](#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](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#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](https://github.com/Elijas/redis-message-queue/blob/main/docs/operations.md#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](https://github.com/Elijas/redis-message-queue/blob/main/docs/operations.md#known-limitations).

> **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](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#graceful-shutdown) and
> [Abandoned in-flight messages](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#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](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md)**:

- **[Deduplication](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#deduplication)** — publish-side dedup keys, TTL windows, and cardinality guidance
- **[Success and failure tracking](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#success-and-failure-tracking)** — optional completed/failed audit lists and their caps
- **[Publish backpressure](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#publish-backpressure)** — `max_pending_length` and the `raise` / `drop_oldest` / `block` overload policies
- **[Crash recovery with visibility timeout](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#crash-recovery-with-visibility-timeout)** — leases, heartbeats, and redelivery
- **[Ordering and multi-consumer fairness](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#ordering-and-multi-consumer-fairness)** — the claim-order guarantee and its limits
- **[Dead-letter queue](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#dead-letter-queue)** — routing poison messages off the redelivery path
- **[Graceful shutdown](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#graceful-shutdown)** — `drain()` and the three shutdown shapes
- **[Custom gateway](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#custom-gateway)** — tuning retries and dedup TTL, or subclassing for new semantics
- **[Connection pool sizing](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#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](https://github.com/Elijas/redis-message-queue/blob/main/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`):

```python
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](https://github.com/Elijas/redis-message-queue/blob/main/docs/api-reference.md) 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](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#graceful-shutdown) and
  [Abandoned in-flight messages](https://github.com/Elijas/redis-message-queue/blob/main/docs/configuration.md#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](https://github.com/Elijas/redis-message-queue/blob/main/docs/operations.md)** for [fork safety and pre-fork servers](https://github.com/Elijas/redis-message-queue/blob/main/docs/operations.md#fork-safety-and-pre-fork-servers) (gunicorn `--preload`, `multiprocessing`, `ProcessPoolExecutor`) and [Redis memory sizing for deduplication and replay metadata](https://github.com/Elijas/redis-message-queue/blob/main/docs/operations.md#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](https://github.com/Elijas/redis-message-queue/blob/main/docs/operations.md#inspecting-and-managing-queues) and the [Redis key layout reference](https://github.com/Elijas/redis-message-queue/blob/main/docs/operations.md#redis-key-layout).

### Production patterns

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

- [`backpressure.py`](https://github.com/Elijas/redis-message-queue/blob/main/examples/production/backpressure.py) ([async](https://github.com/Elijas/redis-message-queue/blob/main/examples/production/asyncio/backpressure.py)) — publish under a bounded pending queue, retrying on `QueueBackpressureError` instead of growing memory unbounded.
- [`graceful_shutdown.py`](https://github.com/Elijas/redis-message-queue/blob/main/examples/production/graceful_shutdown.py) ([async](https://github.com/Elijas/redis-message-queue/blob/main/examples/production/asyncio/graceful_shutdown.py)) — finish the in-flight message and stop cleanly on SIGINT/SIGTERM.
- [`observability.py`](https://github.com/Elijas/redis-message-queue/blob/main/examples/production/observability.py) ([async](https://github.com/Elijas/redis-message-queue/blob/main/examples/production/asyncio/observability.py)) — wire the `on_event` callback to metrics/logs with secret-safe `event.error` handling.
- [`idempotent_consumer.py`](https://github.com/Elijas/redis-message-queue/blob/main/examples/production/idempotent_consumer.py) ([async](https://github.com/Elijas/redis-message-queue/blob/main/examples/production/asyncio/idempotent_consumer.py)) — 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](https://github.com/Elijas/redis-message-queue/blob/main/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](https://github.com/Elijas/redis-message-queue/blob/main/docs/operations.md#known-limitations)**. For the full residual-risk register, see **[docs/production-readiness.md](https://github.com/Elijas/redis-message-queue/blob/main/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](https://github.com/Elijas/redis-message-queue/blob/main/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](https://github.com/Elijas/redis-message-queue/blob/main/UPGRADING.md)**. Per-release detail lives in **[CHANGELOG.md](https://github.com/Elijas/redis-message-queue/blob/main/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:

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

Try the [examples](https://github.com/Elijas/redis-message-queue/tree/main/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.

```bash
# 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.

## License

Released under the [MIT License](https://github.com/Elijas/redis-message-queue/blob/main/LICENSE).

![GitHub Repo stars](https://img.shields.io/github/stars/elijas/redis-message-queue?style=flat&color=fcfcfc&labelColor=white&logo=github&logoColor=black&label=stars)
