Python client

The PyPI package is queen-mq; the import name is queen. It is fully async, built on httpx, and ships a py.typed marker. Version 1.0.0, aligned with the broker.

from queen import Queen, Admin, Stream, token_bucket_gate, sliding_window_gate

The package also exports the typed dicts Message, AckResponse, BufferStats, DLQResponse, TransactionResponse, and the default dicts CLIENT_DEFAULTS, QUEUE_DEFAULTS, CONSUME_DEFAULTS, POP_DEFAULTS, BUFFER_DEFAULTS.

Constructor

Queen(config=None, *, urls=None, url=None, timeout_millis=None, retry_attempts=None,
      retry_delay_millis=None, load_balancing_strategy=None, affinity_hash_ring=None,
      enable_failover=None, health_retry_after_millis=None, bearer_token=None,
      headers=None, retry_429=None, transport=None)

config may be a URL string, a list of URL strings, or a dict. Keyword arguments override whatever config supplied; a None keyword is ignored rather than treated as an override. URLs are validated. Neither url nor urls raises ValueError("Must provide urls or url in configuration").

The load-balancing arguments are inert with a single URL: the LoadBalancer is only constructed when urls has more than one entry.

retry_attempts and failover are mutually exclusive paths. With one URL, or with enable_failover=False, a call makes up to retry_attempts attempts against the same backend (3 by default). With several URLs and failover on, retry_attempts is not consulted at all: the call walks the backend list, trying each at most once.

retry_429

A plain dict; see queen.types.Retry429Config. Separate from retry_attempts, which covers 5xx and network faults.

A Retry-After header (seconds) wins over the computed delay. Every delay gets ±20% jitter. A 429 is retried against the same backend and never triggers failover.

Queen

Queen is an async context manager: async with Queen(url) as client: closes on exit.

ack(message, status, context)

message is a message dict, a transaction-id string, or a list of either. status is True → completed, False → failed, or a status string passed through verbatim. context accepts {"group": ..., "error": ...}.

partitionId is mandatory on every message dict. Without it a single ack returns {"success": False, "error": "Message must have partitionId property to ensure message uniqueness"} and a batch raises ValueError. A leaseId on the message is forwarded when present. Per-message statuses inside a batch work by tagging items with _status and _error; if any item carries either key the whole batch switches to per-item mode.

Return shapes differ between the two paths:

renew(message_or_lease_id)

Accepts a lease-id string, a message dict, or a list of either. Lease ids are deduplicated in insertion order: with multi-partition pop, every message in a batch shares one leaseId and one extend call renews every claimed partition, so passing the whole list issues one HTTP call.

QueueBuilder

client.queue(name) returns a fresh builder. Configuration methods return self; terminal methods are push(), pop(), consume(), create(), delete(), dlq() and flush_buffer().

Addressing

Queue lifecycle

QUEUE_DEFAULTS uses snake_case keys and the builder converts them to the wire’s camelCase before sending:

Any extra key you pass to config() is converted the same way and sent, so options absent from the defaults dict are still reachable. /configure is a full replace: keys absent from the body reset to the broker’s defaults, so send the complete intended configuration every time. The full option list is in Reference.

Producing

Each item may carry data, payload, or be used as the payload itself. The client mints a UUIDv7 transactionId when the item does not supply one, and forwards traceId only when it is a valid UUID.

res = await client.queue("test-queue-v2").push([{"data": {"message": "Hello, world!"}}])

PushBuilder

Awaitable: await executes the push. on_success(cb), on_error(cb) and on_duplicate(cb) register callbacks and return the builder. Awaiting a non-buffered push resolves to the broker’s per-item array; a buffered push resolves immediately to a buffered marker without sending anything.

Consuming

pop()

async pop() → list[dict], [] when there is nothing to return.

res = await client.queue("test-queue-v2-pop-non-empty").batch(1).wait(True).pop()

Route selection: queue plus non-Default partition uses /api/v1/pop/queue/:queue/partition/:partition; queue alone uses /api/v1/pop/queue/:queue; namespace or task without a queue uses /api/v1/pop. None of the three raises ValueError.

consume(handler, *, signal=None)

Returns a ConsumeBuilder, awaitable. signal is an asyncio.Event; setting it stops every worker at the top of its next iteration. on_success(cb) and on_error(cb) register callbacks.

await client.queue("orders").group("workers").batch(20).each().consume(handle)

Worker loop behaviour:

• A handler exception with auto_ack on nacks the message and does not re-raise, so the worker continues.
• In each() mode a nack abandons the rest of the popped batch: the nack released the lease and clamped the broker cursor, so the remainder will be redelivered anyway.
• A 429 that escapes the HTTP layer’s own retry sleeps for Retry-After (or 1 s) and continues.
• A network error sleeps and continues.
• A 403 stops the worker and re-raises.

message["trace"]

Messages delivered through consume() gain a trace coroutine that posts to /api/v1/traces:

await msg["trace"]({"traceName": ["tenant-acme"], "eventType": "info", "data": {"step": 1}})

data is required, traceName accepts a string or list of strings, eventType defaults to "info". It never raises; failures return {"success": False, "error": ...}.

Buffering

A buffer flushes at message_count messages or time_millis after its first message, whichever comes first. Buffers are keyed on "<queue>/<partition>", so builders addressing the same queue and partition share one.

Dead-letter queue

dlq(consumer_group=None) returns a DLQBuilder. Raises ValueError with no queue name.

Read-only. Replay lives on the admin façade: admin.retry_message(partition_id, transaction_id).

TransactionBuilder

client.transaction(). Pushes and acks in one PostgreSQL transaction, all-or-nothing.

await (client.transaction()
       .ack(msg)
       .queue("orders.enriched").push([{"data": {"id": 1}}])
       .commit())

Admin

client.admin (property) or Admin(http_client). Every method is async and returns the broker’s JSON body verbatim. Methods taking **params build a query string, dropping None.

Resources

Messages and traces

Status and analytics

Consumer groups

seek_consumer_group posts options verbatim: {"toEnd": True} or {"timestamp": "<RFC3339>"}.

System

Streaming SDK

Stream builds an immutable operator chain over a queue and runs it against /streams/v1/*. Every combinator returns a new Stream.

Stream.from_(client.queue("orders"), **options)   # or Stream.from_queue(...)

Window keyword arguments mirror the JavaScript names in snake_case and are converted before they reach the wire. token_bucket_gate(...) and sliding_window_gate(...) return ready-made gate functions.

run(query_id, url, **opts)

async, returns a Runner. query_id and url are positional-or-keyword and both required. An empty query_id raises ValueError("run requires query_id=...").

Options are passed as snake_case and mapped onto the runner’s camelCase keys internally. The chain shape is fingerprinted into a config hash; re-deploying a different chain under the same query_id is rejected at registration unless reset=True. Only operator kinds and their structural config are hashed, not the bodies of your functions.

Runner exposes await stop() and metrics().

Logging

Off unless QUEEN_CLIENT_LOG=true is set in the environment. Levels are emitted through the module logger in queen.utils.logger.

Errors

HTTP failures raise httpx exceptions; the client attaches the parsed body’s code to the exception when present. The proxy codes are rate_limited and quota_exceeded on 429, and cluster_suspended, storage_quota_exceeded, feature_gated, forbidden on 403. A 204 response yields None: the broker sends no body at all on 204.

Same surface, other languages