queen-mq 0.12.2
High-performance message queue client for Python backed by PostgreSQL
License: Apache-2.0
Requires Python: >=3.8
Keywords: message-queue, queue, broker, message-queue-system, fifo

# Queen MQ - Python Client

<div align="center">

**Modern, high-performance message queue client for Python**

[![PyPI](https://img.shields.io/pypi/v/queen-mq.svg)](https://pypi.org/project/queen-mq/)
[![License](https://img.shields.io/badge/license-Apache%202.0-blue.svg)](LICENSE)
[![Python](https://img.shields.io/badge/python-3.8%2B-brightgreen.svg)](https://www.python.org/)

[Quick Start](#quick-start) • [Features](#features) • [Documentation](#documentation) • [Examples](#examples)

</div>

---

## What is Queen MQ?

Queen MQ is a PostgreSQL-backed message queue system with a powerful feature set:

- **FIFO Partitions** - Unlimited ordered partitions within queues
- **Consumer Groups** - Kafka-style consumer groups for scalability
- **Flexible Semantics** - Exactly-once, at-least-once, and at-most-once delivery
- **Transactions** - Atomic operations across push and ack
- **High Performance** - 200K+ messages/sec with proper batching
- **Subscription Modes** - Process from beginning, new messages only, or from timestamp
- **Dead Letter Queue** - Automatic failure handling and monitoring
- **Message Tracing** - Debug distributed workflows with trace timelines
- **Client-Side Buffering** - 10x-100x throughput boost for high-volume pushes
- **Real-time Streaming** - Windowed aggregation and processing

This client provides a fluent, async/await API for Python applications.

---

## Installation

```bash
pip install queen-mq
```

**Requirements:** Python 3.8+

---

## Quick Start

```python
import asyncio
from queen import Queen

async def main():
    # Connect to Queen server
    async with Queen('http://localhost:6632') as queen:
        # Create a queue
        await queen.queue('tasks').create()
        
        # Push messages
        await queen.queue('tasks').push([
            {'data': {'task': 'send-email', 'to': 'alice@example.com'}}
        ])
        
        # Consume messages
        await queen.queue('tasks').consume(async def process(message):
            print('Processing:', message['data'])
            # Auto-ack on success, auto-retry on error
        )

asyncio.run(main())
```

---

## Core Concepts

### Queues

Logical containers for messages with configurable settings:

```python
await queen.queue('orders').config({
    'leaseTime': 300,          # 5 minutes
    'retryLimit': 3,
    'priority': 5,
    'encryptionEnabled': False
}).create()
```

### Partitions

Ordered lanes within a queue:

```python
# All messages for user-123 are processed in order
await queen.queue('user-events').partition('user-123').push([
    {'data': {'event': 'login'}},
    {'data': {'event': 'view-page'}},
    {'data': {'event': 'logout'}}
])
```

### Consumer Groups

Multiple consumers sharing work:

```python
# Worker 1 & 2 share the load
await queen.queue('emails').group('processors').consume(async def handler(msg):
    await send_email(msg['data'])
)

# Separate group processes same messages independently
await queen.queue('emails').group('analytics').consume(async def handler(msg):
    await log_metrics(msg['data'])
)
```

### Subscription Modes

Control whether consumer groups process historical messages:

```python
# Default: Process ALL messages (including backlog)
await queen.queue('events').group('batch-analytics').consume(handler)

# Skip history, only new messages
await queen.queue('events').group('realtime-monitor').subscription_mode('new').consume(handler)

# Start from specific timestamp
await queen.queue('events').group('replay').subscription_from('2025-10-28T10:00:00.000Z').consume(handler)
```

---

## Connection Options

### Single Server

```python
queen = Queen('http://localhost:6632')
```

### Multiple Servers (High Availability)

```python
queen = Queen(['http://server1:6632', 'http://server2:6632'])
```

### Full Configuration

```python
queen = Queen({
    'urls': ['http://server1:6632', 'http://server2:6632'],
    'timeout_millis': 30000,
    'retry_attempts': 3,
    'load_balancing_strategy': 'affinity',  # or 'round-robin', 'session'
    'enable_failover': True
})
```

---

## Basic Usage Patterns

### Push Messages

```python
# Simple push
await queen.queue('tasks').push([
    {'data': {'job': 'resize-image', 'imageId': 123}}
])

# With partition
await queen.queue('tasks').partition('tenant-456').push([
    {'data': {'action': 'process'}}
])

# With custom transaction ID (for exactly-once)
await queen.queue('tasks').push([
    {'transactionId': 'unique-id-123', 'data': {'value': 42}}
])
```

### Consume Messages (Long-Running Workers)

```python
# Single message processing (batch=1, default)
# Handler receives a single message
await queen.queue('tasks').concurrency(10).consume(async def handler(message):
    await process_task(message['data'])
    # Auto-ack on success, auto-retry on error
)

# Batch processing (batch>1)
# Handler receives an array of messages
await queen.queue('tasks').batch(20).concurrency(5).consume(async def handler(messages):
    for message in messages:
        await process_task(message['data'])
)

# Process with limit and stop
await queen.queue('tasks').limit(100).consume(async def handler(message):
    await process_task(message['data'])
)
```

### Pop Messages (On-Demand Processing)

```python
# Grab messages manually
messages = await queen.queue('tasks').batch(10).wait(True).pop()

# Manual acknowledgment
for message in messages:
    try:
        await process_message(message['data'])
        await queen.ack(message, True)  # Success
    except Exception as error:
        await queen.ack(message, False)  # Retry
```

### Transactions (Atomic Operations)

```python
# Pop from queue A
messages = await queen.queue('input').pop()

# Atomically: ack input AND push output
await (queen.transaction()
    .ack(messages[0])
    .queue('output')
    .push([{'data': processed_result}])
    .commit())
```

### Client-Side Buffering (High Throughput)

```python
# Buffer messages locally, batch to server
for i in range(10000):
    await queen.queue('events').buffer({'message_count': 500, 'time_millis': 1000}).push([
        {'data': {'id': i}}
    ])

# Flush remaining buffered messages
await queen.flush_all_buffers()

# Result: 10x-100x faster than individual pushes
```

### Dead Letter Queue

```python
# Enable DLQ on queue
await queen.queue('risky').config({'retryLimit': 3, 'dlqAfterMaxRetries': True}).create()

# Query failed messages
dlq = await queen.queue('risky').dlq().limit(10).get()

print(f"Found {dlq['total']} failed messages")
for msg in dlq['messages']:
    print('Error:', msg.get('errorMessage'))
```

### Message Tracing

```python
await queen.queue('orders').consume(async def handler(msg):
    order_id = msg['data']['orderId']
    
    # Record trace with name for cross-service correlation
    await msg['trace']({
        'traceName': f"order-{order_id}",
        'eventType': 'info',
        'data': {'text': 'Order processing started'}
    })
    
    await process_order(msg['data'])
    
    await msg['trace']({
        'traceName': f"order-{order_id}",
        'eventType': 'processing',
        'data': {'text': 'Order completed', 'total': msg['data']['total']}
    })
)

# View traces in webapp: Traces → Search "order-12345"
```

---

## API Reference

### Queue Operations

```python
# Create
await queen.queue('my-queue').create()
await queen.queue('my-queue').config({'priority': 5}).create()

# Delete
await queen.queue('my-queue').delete()
```

### Push

```python
await queen.queue('q').push([{'data': {'value': 1}}])
await queen.queue('q').partition('p1').push([{'data': {'value': 1}}])
await queen.queue('q').buffer({'message_count': 100, 'time_millis': 1000}).push([...])
```

### Pop

```python
msgs = await queen.queue('q').pop()
msgs = await queen.queue('q').batch(10).pop()
msgs = await queen.queue('q').batch(10).wait(True).pop()
```

### Consume

```python
# batch=1 (default): handler receives single message
await queen.queue('q').consume(async def handler(msg): ...)

# batch>1: handler receives array of messages
await queen.queue('q').batch(10).consume(async def handler(msgs): ...)

# Other options
await queen.queue('q').limit(10).consume(handler)
await queen.queue('q').concurrency(5).consume(handler)
await queen.queue('q').group('my-group').consume(handler)
```

### Acknowledgment

```python
await queen.ack(message, True)   # Success
await queen.ack(message, False)  # Retry
await queen.ack(message, False, {'error': 'reason'})
await queen.ack([msg1, msg2], True)  # Batch ack
```

### Transactions

```python
await (queen.transaction()
    .ack(message)
    .queue('output')
    .push([{'data': {'result': 'processed'}}])
    .commit())
```

### Lease Renewal

```python
await queen.renew(message)
await queen.renew([msg1, msg2, msg3])
await queen.queue('q').renew_lease(True, 60000).consume(handler)
```

### Buffering

```python
await queen.flush_all_buffers()
await queen.queue('q').flush_buffer()
stats = queen.get_buffer_stats()
```

### Dead Letter Queue

```python
dlq = await queen.queue('q').dlq().limit(10).get()
dlq = await queen.queue('q').dlq('consumer-group').limit(10).get()
```

### Shutdown

```python
await queen.close()  # Flush buffers and close connections
```

---

## Configuration Defaults

### Client Defaults

```python
{
    'timeout_millis': 30000,
    'retry_attempts': 3,
    'retry_delay_millis': 1000,
    'load_balancing_strategy': 'affinity',
    'enable_failover': True
}
```

### Queue Defaults

```python
{
    'leaseTime': 300,           # 5 minutes
    'retryLimit': 3,
    'priority': 0,
    'delayedProcessing': 0,
    'windowBuffer': 0,
    'maxSize': 0,              # Unlimited
    'retentionSeconds': 0,     # Keep forever
    'encryptionEnabled': False
}
```

### Consume Defaults

```python
{
    'concurrency': 1,
    'batch': 1,
    'auto_ack': True,
    'wait': True,              # Long polling
    'timeout_millis': 30000,
    'limit': None,             # Run forever
    'renew_lease': False
}
```

---

## Logging

Enable detailed logging for debugging:

```bash
export QUEEN_CLIENT_LOG=true
python your_app.py
```

Example output:
```
[2025-10-28T10:30:45.123Z] [INFO] [Queen.constructor] {"status":"initialized","urls":1}
[2025-10-28T10:30:45.234Z] [INFO] [QueueBuilder.push] {"queue":"tasks","partition":"Default","count":5}
```

---

## Type Hints

Full type hints included for IDE support:

```python
from queen import Queen, Message
from typing import Dict, Any

queen: Queen = Queen('http://localhost:6632')

async def handler(message: Message) -> None:
    data: Dict[str, Any] = message['data']
    print(data)

await queen.queue('orders').consume(handler)
```

---

## Best Practices

1. ✅ **Use `consume()` for workers** - Simpler API, handles retries automatically
2. ✅ **Use `pop()` for control** - When you need precise control over acking
3. ✅ **Buffer for speed** - Always use buffering when pushing many messages
4. ✅ **Partitions for order** - Use partitions when message order matters
5. ✅ **Consumer groups for scale** - Run multiple workers in the same group
6. ✅ **Transactions for consistency** - Use transactions for atomic operations
7. ✅ **Enable DLQ** - Always enable DLQ in production
8. ✅ **Renew long leases** - Use auto-renewal for long-running tasks
9. ✅ **Graceful shutdown** - Use async context manager or call `queen.close()`
10. ✅ **Monitor DLQ** - Regularly check for failed messages

### 📝 Important Notes

**Handler Signatures:**
- When `batch=1` (default), handler receives a **single message**: `async def handler(message): ...`
- When `batch>1`, handler receives an **array of messages**: `async def handler(messages): ...`
- When `each=True`, always receives single messages regardless of batch size

---

## Documentation

- **[Node.js Client](../client-js/README.md)** - Node.js client documentation
- **[HTTP API Reference](https://github.com/smartpricing/queen/blob/master/server/API.md)** - Raw HTTP endpoints
- **[Server Guide](https://github.com/smartpricing/queen/blob/master/server/README.md)** - Server setup and configuration
- **[Architecture Guide](https://github.com/smartpricing/queen/blob/master/documentation/ARCHITECTURE.md)** - Deep dive into internals

---

## License

Apache 2.0 - See [LICENSE](../LICENSE.md)

---

## Support

- **GitHub:** [smartpricing/queen](https://github.com/smartpricing/queen)
- **Issues:** [GitHub Issues](https://github.com/smartpricing/queen/issues)
- **LinkedIn:** [Smartness](https://www.linkedin.com/company/smartness-com/)