Metadata-Version: 2.4
Name: queen-mq
Version: 1.2.0
Summary: High-performance message queue client + fluent streaming SDK for Python, backed by PostgreSQL
Author-email: Smartness <info@smartness.com>
License: Apache-2.0
Project-URL: Homepage, https://queenmq.com
Project-URL: Documentation, https://queenmq.com/use/
Project-URL: Repository, https://github.com/queen-mq/queen
Project-URL: Issues, https://github.com/queen-mq/queen/issues
Keywords: message-queue,queue,broker,message-queue-system,fifo,streaming,stream-processing,rate-limiter,tumbling-window,sliding-window,session-window,event-time,watermark,exactly-once,postgres,postgresql
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.27.0
Requires-Dist: typing-extensions>=4.0.0; python_version < "3.10"
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: asyncpg>=0.29.0; extra == "dev"
Requires-Dist: black>=23.0.0; extra == "dev"
Requires-Dist: mypy>=1.0.0; extra == "dev"
Requires-Dist: ruff>=0.1.0; extra == "dev"
Dynamic: license-file

# 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** — 104K msg/s push, 165K msg/s fan-out with consumer groups on a single 32-core node ([benchmarks](https://github.com/queen-mq/queen/tree/master/benchmark-queen/2026-04-26))
- **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 process(message):
    print('Processing:', message['data'])
    # Auto-ack on success, auto-retry on error

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 (handler is a regular async function)
        await queen.queue('tasks').consume(process)

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
async def send_handler(msg):
    await send_email(msg['data'])

async def analytics_handler(msg):
    await log_metrics(msg['data'])

# Worker 1 & 2 share the load
await queen.queue('emails').group('processors').consume(send_handler)

# Separate group processes same messages independently
await queen.queue('emails').group('analytics').consume(analytics_handler)
```

### Subscription Modes

Control whether consumer groups process historical messages:

```python
async def handler(msg):
    ...

# 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
async def single_handler(message):
    await process_task(message['data'])
    # Auto-ack on success, auto-retry on error

await queen.queue('tasks').concurrency(10).consume(single_handler)

# Batch processing (batch>1)
# Handler receives an array of messages
async def batch_handler(messages):
    for message in messages:
        await process_task(message['data'])

await queen.queue('tasks').batch(20).concurrency(5).consume(batch_handler)

# Process with limit and stop
await queen.queue('tasks').limit(100).consume(single_handler)
```

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

### Multi-Partition Pop (Drain Many Partitions Per Call)

```python
# One round-trip drains up to 200 messages spread across up to 50 partitions.
# batch(200) is the GLOBAL cap on total messages; partitions(50) is the
# hard cap on partitions claimed. All claimed partitions share one leaseId
# — a single renew() call extends every partition's lease atomically.
messages = await (queen.queue('events')
                  .batch(200)
                  .partitions(50)
                  .wait(True)
                  .pop())

# Each message carries its own partition info (per-message partitionId,
# partition name, leaseId, consumerGroup) — ACK and renew always work
# message-by-message regardless of how many partitions the batch spans.
for m in messages:
    print(f"from {m['partition']}:", m['data'])

# Same builder works on .consume() for long-running workers
async def handler(msgs):
    for m in msgs:
        await process(m['data'])

await (queen.queue('events')
       .batch(100)
       .partitions(8)
       .consume(handler))
```

**When to use:** queues with many partitions where each partition only has
a handful of new messages per polling interval (per-customer event streams,
per-tenant work queues, per-device telemetry). Reduces network round-trips
from O(P) to O(P / N) while preserving per-partition FIFO ordering.

**When not to use:** few partitions, or each one busy enough to fill
`batch(B)` on its own. Leaving `.partitions()` unset hands the sweep width to
the broker (see Pop Autopilot below); `.partitions(1)` pins the legacy
single-partition behaviour.

`.partitions(N)` only applies to **wildcard** pops; specifying
`.partition('name')` ignores the cap.

### Pop Autopilot (Let the Broker Size the Pop)

Since 1.2, `batch` and `partitions` that you do **not** set are chosen by the
broker, per pop, from state the client cannot see: how many partitions of the
group are ready, how old their oldest ready message is, how fast messages are
arriving. The knobs you *do* set are never touched.

```python
# Both knobs are the broker's: it picks the sweep width and the budget.
await (queen.queue('events').group('workers')
       .consume(handler))

# One knob pinned, one delegated: this consumer stays on one partition
# forever, and the broker sizes the batch for it.
await (queen.queue('events').group('workers').partitions(1)
       .consume(handler))
```

The request carries `autopilot=true` and simply omits the delegated knobs.
Setting both leaves nothing to decide, so nothing changes on the wire at all.

**Two ways to switch it off**, both restoring the previous client-side
defaults (batch 1, partitions 1) byte for byte:

```python
await queen.queue('events').autopilot(False).consume(handler)
```

```bash
QUEEN_SDK_POP_AUTOPILOT=off   # whole process, read once at client creation
```

**What the broker chose** rides back on the response and is there for the
reading, along with an optional pacing hint the consume loop honours in place
of its own delay between empty polls:

```python
res = await queen.queue('events').group('workers').pop_result()
if res.autopilot:
    print(f'{res.autopilot.partitions} partitions, batch {res.autopilot.batch}, '
          f'poll again in {res.autopilot.wait_millis}ms')
```

**Requires broker >= 1.2.** An older broker ignores the parameter, so the
omitted knobs take *its* defaults (batch 200, partitions 1) instead of the old
client-side ones. That is a sizing difference and nothing else — no message is
lost, reordered or duplicated — so unlike conflation it degrades silently and
on purpose. Pin the values explicitly, or turn autopilot off, if you need the
old numbers against an old broker.

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

Buffer options:

| Option | Default | Meaning |
| --- | --- | --- |
| `message_count` | 100 | Flush once this many messages are buffered (also the batch size of one POST) |
| `time_millis` | 1000 | Flush this long after the first buffered message |
| `max_size` | `4 * message_count` | Backpressure bound: `push()` BLOCKS while this many messages are waiting |
| `retry_delay_millis` | 250 | Wait before retrying a batch whose POST failed |

The buffer is bounded and `push()` blocks at the bound, so a producer that
outruns the flush pipeline is slowed down to the drain rate instead of growing
the process until it dies with every unflushed message inside it. `max_size` of
0 means the default bound, not unbounded. A buffered `push()` therefore either
returns `{'buffered': True, ...}` or raises: cancel it (`asyncio.wait_for`) if
your producer needs a deadline, and treat the exception as "not buffered".

A batch whose POST fails goes back to the front of the buffer and is retried
after `retry_delay_millis`, in order, until it lands. Nothing is dropped, so a
broker outage shows up as blocked producers rather than as missing messages.
`flush_all_buffers()` keeps retrying while the broker is down (cancel it if you
need a bounded wait); `close()` already bounds it at 30 seconds and logs how
many messages were left unsent, so a shutdown ends loudly instead of hanging.

### 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
async def order_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']}
    })

await queen.queue('orders').consume(order_handler)

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

---

## Key/Value State and Timers

Both surfaces are **always there**. There is no flag to turn on before using
them and nothing to probe first: any broker you can push to can also hold state
and schedule timers.

An operator can still pause either one during an incident, in which case a call
raises `KvError` / `TimerError` with `status == 503`, `code` `kv_disabled` or
`timers_disabled`, and a `Retry-After` in `retry_after_seconds`. Inside a
transaction the same pause is a `403` instead, so a bundle fails fast rather
than spinning with messages in hand. Timer *cancels* are never paused.

### Key/Value

```python
from datetime import timedelta

# An expiry is MANDATORY on every write: exactly one of ttl_seconds, ttl,
# until or forever=True. A put never inherits the previous expiry: that is
# the fastest way to make a marker immortal.
await queen.kv.put('orders', 'order:9f1', {'state': 'held'}, ttl_seconds=60)
await queen.kv.put('orders', 'order:9f1', {'state': 'held'}, ttl=timedelta(minutes=1))

got = await queen.kv.get('orders', 'order:9f1')
if got:                       # follows `found`, not the value
    print(got['value'], got['version'])

# "Did I win?", in one call, one boolean. This is the idempotency marker.
if await queen.kv.once('dedup', f"evt:{event_id}", ttl_seconds=86400):
    await do_the_external_effect()

# Optimistic lock. expect=0 means "must not exist"; expect=N is a pure update
# that creates nothing when it matches no row.
res = await queen.kv.put('orders', 'order:9f1', {'state': 'shipped'},
                         ttl_seconds=60, expect=got['version'])
if not res:
    print(res['reason'])      # 'version' | 'absent' | 'exists' | 'limit' | 'type'

# Rate limiting without a CAS loop. With max, `applied` IS the admission
# decision: nothing saturates, nothing truncates, a refusal spends no budget.
allowed = await queen.kv.incr('quota', f"{customer}:{hour}", delta=1, max=1000, ttl_seconds=3600)
if not allowed:
    raise TooManyRequests()

rows = await queen.kv.list_all('saga', 'order:9f1:')   # follows nextAfter
```

**A write that did not apply is not an error.** `applied: false` answers HTTP
200 with the current value and version, so the loser needs no second round
trip. Results are falsy when they did not apply, so `if await
queen.kv.delete(...)` reads the verdict rather than "an object came back".

**Read-modify-write across two calls is safe only when the KV key derives from
the partition key.** Otherwise the lanes do not serialise it for you: use
`incr`, or carry `expect`.

**`put_if_absent` plus a TTL is not a distributed lock.** A lock that expires is
not revoked: the old holder keeps working, it simply no longer has the row.
Carry your `version` as `expect` on every later write so a lapsed holder fails
with `reason: "version"` instead of overwriting the new one.

### Timers

```python
res = await queen.timers.schedule('orders', 'order:9f1:expire',
                                  {'orderId': '9f1'}, delay_ms=30_000)
res['txn'], res['messageId'], res['deliverAt']

await (queen.timer('orders')
            .key('order:9f1:expire')
            .payload({'orderId': '9f1'})
            .after(timedelta(minutes=30))
            .schedule())

cancelled = await queen.timers.cancel('orders', 'order:9f1:expire', txn=res['txn'])
```

Durations that can be sub-second are in **milliseconds** (`delay_ms`), the ones
that cannot are in **seconds** (`ttl_seconds`). Only relative delays exist:
there is one clock and it is the database's. A delay in the past is legal and
fires on the first cycle.

`deliverAt` is **"not before"**, never "exactly at".

**`absent` means "no longer pending" and may mean ALREADY DELIVERED.** There is
no tombstone: a delivered timer has no row. The response echoes the `txn` back
so the authority, the log, can be consulted without a second API call. A saga
that cancels a compensation timer must therefore have its compensation consumer
re-check the saga state before compensating, because the cancel may have arrived
5 ms after the fire.

Use `queen.timers.cancel(...)`, not a `cancel` op inside a batch: the DELETE
route it takes is the one that is never blocked by a quota. A tenant that cannot
cancel keeps producing messages it cannot stop.

### Inside a transaction

The transaction is the **primary fence**; `expect` is only the secondary
assertion. A state write that shares the transaction with its ack is undone when
an expired lease makes the ack fail, which a CAS cannot do.

```python
result = await (queen.transaction()
    .once('dedup', f"evt:{event_id}", ttl_seconds=86400)   # the gate, first
    .queue('orders').push([{'data': {...}}])
    .ack(message)
    .timer('orders').key(f"order:{oid}:expire").payload({'oid': oid}).after_ms(30_000).schedule()
    .commit())

if not result:
    # RETURNED, not raised: a lost gate is the expected outcome of a legitimate
    # redelivery, so it stays out of your retry policy and your error metrics.
    assert result['reason'] == 'kv_precondition'
    print(result['failedIndex'], result['kvReason'], result['version'], result['value'])
```

Everything else still raises. `get_prefix` is not available inside a
transaction: its cost is not bounded by the caller.

---

## 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()                            # broker-sized (see Pop Autopilot)
msgs = await queen.queue('q').batch(10).pop()
msgs = await queen.queue('q').batch(10).wait(True).pop()
msgs = await queen.queue('q').batch(200).partitions(50).pop()  # multi-partition pop
res = await queen.queue('q').pop_result()                      # + what the broker chose
```

### Consume

```python
# batch=1 (default): handler receives single message
async def single_handler(msg):
    ...

await queen.queue('q').consume(single_handler)

# batch>1: handler receives array of messages
async def batch_handler(msgs):
    ...

await queen.queue('q').batch(10).consume(batch_handler)

# Other options
await queen.queue('q').limit(10).consume(single_handler)
await queen.queue('q').concurrency(5).consume(single_handler)
await queen.queue('q').group('my-group').consume(single_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])

async def handler(msg):
    ...
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,                # autopilot OFF only -- unset means the broker sizes it
    'max_partitions': 1,       # autopilot OFF only -- unset means the broker sizes it
    'auto_ack': True,
    'wait': True,              # Long polling
    'timeout_millis': 30000,
    'limit': None,             # Run forever
    'renew_lease': False
}
```

`batch` and `max_partitions` are the **autopilot-off** defaults: with autopilot
on (the default) a knob you never set is not defaulted at all, it is delegated
to the broker. These values are what comes back with `.autopilot(False)` or
`QUEEN_SDK_POP_AUTOPILOT=off`.

---

## 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/queen-mq/queen/blob/master/server/API.md)** — raw HTTP endpoints
- **[Server Guide](https://github.com/queen-mq/queen/blob/master/server/README.md)** — server setup and configuration
- **[Architecture & internals](https://queenmq.com/architecture.html)** — published architecture overview
- **[libqueen design notes](https://github.com/queen-mq/queen/blob/master/cdocs/LIBQUEEN_IMPROVEMENTS.md)** — adaptive engine deep-dive

---

## License

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

---

## Support

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

