Metadata-Version: 2.4
Name: redis-queue-broker
Version: 1.0.4
Summary: A simple Redis-queue message broker for python projects
Author-email: Andrew Avramenko <nallanor@gmail.com>
License: MIT
Project-URL: Homepage, https://github.com/nallan-dev/redis-queue-broker
Project-URL: Repository, https://github.com/nallan-dev/redis-queue-broker
Keywords: redis,queue,broker
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: 3.15
Classifier: Programming Language :: Python :: 3.16
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENCE
Requires-Dist: redis>=5.0.1
Dynamic: license-file


# Redis Queue Broker

A simple, flexible Redis‑based queue broker for Python that supports both synchronous and asynchronous usage via the official `redis` library.

---

## Installation

```bash
pip install redis-queue-broker
````
The package depends on redis>=5.0.1.

## Quick start
1) Define a queue configuration

```python
from redis_queue_broker import BaseRedisQueueBroker

class MyQueue(BaseRedisQueueBroker):
    QUEUE_NAME = "my_queue"
    QUEUE_MAX_SIZE = 10  # (FIFO is True by default)

class MyAnotherQueue(BaseRedisQueueBroker):
    QUEUE_NAME = "my_another_queue"
    QUEUE_MAX_SIZE = 10
    FIFO = False  # LIFO now
```

2.1. Producer (for synchronous Redis example)

```python
from redis import Redis

from somwhere import MyQueue

# We initialize MyQueue with synchronous Redis, 
#  so our broker instance also becomes synchronous
sync_redis = Redis.from_url("redis://localhost:6379")
broker = MyQueue(redis=sync_redis)
broker.send("Hello, Redis!", rewrite=False)  # default rewrite=True
print(broker.get_queue_length())  # 1
```

2.2. Producer (for asynchronous Redis example)

```python
import asyncio
from redis.asyncio.client import Redis as AsyncRedis

from somwhere import MyQueue

# We initialize MyQueue with asynchronous Redis, 
#  so our broker instance also becomes asynchronous
async_redis = AsyncRedis.from_url("redis://localhost:6379")
broker = MyQueue(redis=async_redis)


async def run_sample():
    await broker.send("Hello, Redis!", rewrite=False)  # default rewrite=True
    queue_len = await broker.get_queue_length()  # 1
    print(queue_len)
    
asyncio.run(run_sample())
```

3.1. Consumer (for synchronous Redis example)

```python
from redis import Redis
from redis_queue_broker import ReceiverError, ReceiverEvent

from somwhere import MyQueue

# We initialize MyQueue with synchronous Redis, 
#  so our broker instance also becomes synchronous
sync_redis = Redis.from_url("redis://localhost:6379")
broker = MyQueue(redis=sync_redis)

# Attach synchronous callback
@broker.receiver
def handle_message(event: ReceiverEvent):
    print(f"Received: {event.message.content}")
    
# Attach synchronous error handler (optional)
@broker.error_handler
def handle_error(error: ReceiverError):
    print(f"Error: {error.error}")

broker.listen_forever(workers=3)  # threads
```

3.2. Consumer (for asynchronous Redis example)

```python
import asyncio
from redis.asyncio.client import Redis as AsyncRedis
from redis_queue_broker import ReceiverError, ReceiverEvent

from somwhere import MyQueue

# We initialize MyQueue with asynchronous Redis, 
#  so our broker instance also becomes asynchronous
async_redis = AsyncRedis.from_url("redis://localhost:6379")
broker = MyQueue(redis=async_redis)

# Attach asynchronous callback
@broker.receiver
async def handle_message(event: ReceiverEvent):
    print(f"Received: {event.message.content}")

# Attach asynchronous error handler (optional)
@broker.error_handler
async def handle_error(error: ReceiverError):
    print(f"Error: {error.error}")

asyncio.run(broker.listen_forever(workers=3))  # asyncio-tasks
```
## Features
- FIFO / LIFO – configure with the class attribute FIFO (default True).
- Sync & async – works transparently with both redis.Redis and redis.asyncio.client.Redis.
- Prefixed queue length – limit the queue size; optionally drop oldest items when full.
- Worker pool – spawn multiple concurrent listeners (threads for sync, tasks for async).
- Error handling – attach an optional error handler via the error_handler decorator.
- Read‑only inspection – retrieve all messages without consuming them (get_all_queue_data).
- Original redis-client is accessible via attribute of broker instance.

## API overview
### `BaseRedisQueueBroker`
Abstract base class. Subclass it and set QUEUE_NAME, QUEUE_MAX_SIZE, and optionally FIFO.

| Method / Property             | Description                                                                              |
|-------------------------------|------------------------------------------------------------------------------------------|
| `send(message, rewrite=True)` | Push a message to the queue.                                                             |
| `get_queue_length()`          | Number of items in the queue.                                                            |
| `get_all_queue_data()`        | Return all queue items as `RedisQueueMsg` (read‑only).                                   |
| `listen_forever(workers=1)`   | Block and listen indefinitely, passing each message to the registered receiver.          |
| `receiver(func)`              | Decorator to register a message handler.                                                 |
| `error_handler(func)`         | Decorator to register an error handler (default: wait 5 seconds).                        |
### `ReceiverEvent` (dataclass)
- `message` – `RedisQueueMsg` with `.content` (str) and `.created_at` (datetime).
- `broker` – reference to the broker instance.
- `worker_id` – integer, id of the worker thread/task.
### `ReceiverError` (dataclass)
- `error` – the exception that occurred.
- `broker` – reference to the broker instance.
- `worker_id` – integer, id of the worker thread/task.
### Exceptions
- `QueueFullException` – raised when send() is called with rewrite=False and the queue is full.
- `ImproperlyConfigured` – raised for invalid queue configuration or misuse of decorators.
### License
MIT – see the [LICENCE](LICENCE) file.
