Metadata-Version: 2.4
Name: queen-mq
Version: 2.3.0
Summary: High-performance message queue client + fluent streaming SDK for Python
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
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: zstd
Requires-Dist: zstandard>=0.21.0; python_version < "3.14" and extra == "zstd"
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: zstandard>=0.21.0; python_version < "3.14" and 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 partitioned message queue broker that keeps its state in its own replicated log, 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** — 1M msg/s in and out of one queue of 10M partitions on three nodes ([benchmarks](https://queenmq.com/benchmarks/))
- **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,
    # Opt-in wire forms, off by default (see Binary Push, Push Compression and Framed Pop)
    'binary_push': False,
    'push_compression': False,
    'framed_pop': False
})
```

---

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

### Commit at Delivery

`commit_on_delivery()` commits a `pop()` at delivery: the broker moves the
consumer group's cursor past the messages as it hands them out. There is no
lease and nothing to ack, so the delivery is at-most-once: a message that your
code loses after the pop does not come back. It is a `pop()` option:
`consume()` always leases its messages and raises `ValueError` when it is set.
The broker refuses it together with `conflation()`.

```python
messages = await (queen.queue('metrics')
                  .group('dashboard')
                  .commit_on_delivery()
                  .pop())  # already committed: do not ack these
```

The ephemeral pop has the same option:
`await queen.ephemeral.pop('inbox', commit_on_delivery=True)`. Its old name,
`auto_ack=True`, still works and is deprecated.

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

### Binary Push, Push Compression and Framed Pop

Three opt-in forms of the push, pop and transaction requests. They change how
the messages travel between the client and the broker and nothing else: the
calls, their results, deduplication and the delivery guarantees stay the same.
All three are off by default, and with them off the client sends and reads the
JSON every broker knows.

```python
queen = Queen(
    url='http://localhost:6632',
    binary_push=True,       # a push travels as a binary body
    push_compression=True,  # with binary_push: framed and compressed here
    framed_pop=True,        # a pop returns the stored records, unpacked here
)
```

The same three names are keys of the config dict.

| Option | What it does |
| --- | --- |
| `binary_push` | A push, direct or buffered, sends its items as one length-prefixed binary body instead of a JSON document. The answer is the same list, one entry per item. |
| `push_compression` | Needs `binary_push`. The client frames each partition's batch the way the broker stores it, mints the message ids (UUIDv7) and compresses the batch with zstd, once. The broker checks the frames and, when it has nothing to add to them, stores those bytes without compressing them again. A transaction that pushes sends its pushes the same way, with the rest of the bundle as JSON in front of them. |
| `framed_pop` | `pop()` and `consume()` ask for the stored records (`frames=true`) and unpack them here into the same message dicts, instead of a JSON answer that the broker renders message by message. |

Turn them on when the broker's CPU or the network limits a high message rate:
the framing, the compression and the unpacking then run in the clients, and the
batches cross the network compressed.

`push_compression` and `framed_pop` need zstd. Python 3.14 has it in the
standard library. On an older Python, install the `zstandard` package with the
client:

```bash
pip install "queen-mq[zstd]"
```

Without zstd, `Queen(...)` raises `RuntimeError` when an option that needs it is
on.

**What stays JSON**, with no action from you:

- A push with a `traceId` on one of its items.
- A transaction with no push, or with a `traceId` on one of its pushes.
- A pop the broker cannot answer in frames: a pop of a conflating group, and
  every pop on a broker that runs with an encryption key
  (`QUEEN_ENCRYPTION_KEY`). The client reads the JSON answer as always.
- Acks, KV, timers, locks and the admin calls. Ephemeral queues have their own
  routes and do not change either.

A transaction's `results` come back in the order you added the operations, as
with JSON. The broker answers a binary transaction with the acks first and the
pushes after them, and the client puts them back.

**Requires broker 2.3.0 or later** for `binary_push` and `push_compression`.
A broker up to 2.2.0 answers 400 to the binary bodies and the client does not
fall back to JSON, so turn the two on after every broker of the cluster is
upgraded. `framed_pop` is safe against any broker: an older one ignores the
parameter and answers JSON.

### 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 broker'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.

### Locks and semaphores

A lock is a lease: one holder at a time, for a lifetime the holder renews, with a
token that fences a holder that outlived it. `client.semaphore(name, n, ...)` is the
same with `n` permits.

```python
lock = client.lock("daily-report", ttl_seconds=30)
if not await lock.acquire(wait=5):          # up to 5 s for our turn
    return                                  # somebody else holds it
try:
    await (client.transaction()
           .guard(lock)                     # commits only while the lock is ours
           .queue("reports").push([{"data": report}])
           .commit())
finally:
    await lock.release()

# Or as a context manager, which raises LockNotHeldError when it is held:
async with client.lock("sync:crm", ttl="1m") as lock:
    await sync(lock)
```

It is a lease, not a mutex: it expires, and nobody tells the holder. A paused or
partitioned process carries on past its lifetime while somebody else acquires, so
the lock alone never makes two holders impossible. The guard is what keeps the
old holder's work out: a guarded transaction rolls back with `kv_precondition`
once the lock is no longer this handle's. Outside Queen, fence with the token,
which only rises on a lock. The token changes at every renew, so read it when you
use it.

The handle renews in the background, every third of the lifetime
(`auto_renew=False` to call `lock.renew()` yourself), and says when the lock is
gone: `lock.lost` is an `asyncio.Event` and `lock.on_lost(fn)` runs. Nothing stops
your code: check `lock.lost.is_set()` in a long loop.

`client.locks` is the wire, with no state kept: `acquire`, `renew`, `release`,
`get(name)` and `batch`. `client.kv.check(ns, key, expect=...)` is the precondition
on its own, and `transaction().kv.check(..., required=True)` gates a commit on a key
the transaction does not write.

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

`auto_ack(False)` leaves the ack of a handler that returns to you. A handler
that raises then stops the consumer: `consume()` raises its error and sends no
nack, so the message comes back when its lease expires. With `auto_ack` (the
default), a handler that raises is nacked and the consumer goes on.
`auto_ack()` has no effect on `pop()`: a pop commits at delivery only with
`commit_on_delivery()`.

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

The broker answers HTTP 200 even when it refuses an ack, for example under an
expired lease. `success` is `False` then, with the broker's `error`; for a
batch, each item's verdict is in `results`.

### Transactions

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

### Lease Renewal

```python
res = await queen.renew(message)    # {leaseId, success, newExpiresAt, renewed}
await queen.renew([msg1, msg2, msg3])  # one result per distinct lease

async def handler(msg):
    ...
await queen.queue('q').renew_lease(True, 60000).consume(handler)
```

`success` is `False` when the broker extended nothing: the lease expired, was
released by an ack or nack, or does not exist.

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

To send a message to the DLQ yourself, ack it with the `dlq` status while you
hold its lease: `await queen.ack(message, 'dlq', {'group': group})`.
`admin.move_message_to_dlq()` and `admin.clear_queue()` raise
`NotImplementedError`: the 2.x broker has no route for either.

### 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,
    'binary_push': False,        # a push as a binary body
    'push_compression': False,   # with binary_push: frames compressed in the client
    'framed_pop': False          # a pop as the stored records
}
```

### 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/)


### Optional consumer supervision

```python
await (client.queue("orders").group("billing")
       .supervision({"group": "billing-production"})
       .concurrency(4).each().consume(handler))
```

Supervision defaults to off. `.supervision(None)` or `.supervision(False)` disables
it. The group identifies the application/deployment separately from the consumer
group. Each consume invocation publishes a unique `queen.consumer.status/v1`
instance to the broker's `queen-supervisor` KV namespace every 10 seconds, with a
30-second heartbeat timeout and 60-second TTL, plus stopped status on orderly
exit. The credential needs KV write access. A single asyncio publisher uses the
existing transport and a two-second total deadline; failures do not change ACKs,
lease renewal, handler exceptions or cancellation.

The compatible Supervisors page shows actual live tasks, busy handlers,
successful/failed handler calls, last completion and oldest in-flight duration.
A batch is one call; cancelled handlers count as failed calls. Completion does
not assert ACK success. No message payloads or error text are published.
Event-loop starvation can stop heartbeats; a heartbeat does not prove progress.
Reporting does not restart tasks/processes, autoscale or enable remote commands.
Disabled reporting creates no background task or KV traffic.

#### Match supervision to the consumer lifetime

Each awaited `consume()` execution gets a new instance ID and fresh counters,
even when it reuses the client, builder, queue, hostname and publication group.
A normal exit caused by `.limit(...)` or `.idle_millis(...)` publishes `stopped`
with zero running workers. The final observation has a 60-second TTL; expired
records can remain visible until the broker sweeps them. Repeated short calls
can therefore show many stopped instances for a process whose outer loop is
still running. Compare instance IDs and application logs before diagnosing it.

For a persistent consumer, enable supervision on one execution and omit limits
that deliberately end it. Await its exit under the application's cancellation
policy before closing the client. Several persistent executions in one process
have separate instances and concurrency budgets.

If an outer scheduler rotates queues through short calls, keep per-call
supervision off when you need the scheduler's lifetime status. Publish the
[consumer status contract](https://queenmq.com/reference/supervisor-status/#reporting-an-application-owned-scheduler)
with one ID per scheduler lifetime, stable worker pool names and counters across
turns. The fluent supervision option does not share a reporter or override its
instance ID. Preserve limits used for fairness, the shared worker budget and
existing ACK/NACK, lease renewal and cancellation behavior.
