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.3.0, aligned with the broker’s 1.3 line.

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

retry_429 is a plain dict (see queen.types.Retry429Config) with max_attempts, base_ms and cap_ms, separate from retry_attempts; the defaults, the Retry-After contract and the jitter are in What the clients do with a 429.

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 carries the same nine options as the JavaScript and PHP clients, with the values listed under Queue option defaults, as snake_case keys (lease_time, retry_limit, priority, delayed_processing, window_buffer, max_size, retention_seconds, completed_retention_seconds, encryption_enabled) that the builder converts 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.

Since 1.6.0 /configure merges: an option the body does not carry keeps the value the queue already has, and only "mode": "replace" resets what the body omits. This client sends no mode, so create() always merges. Note what that leaves: because config() merges over QUEUE_DEFAULTS, those nine keys are on the wire whether you named them or not and land on the client’s default rather than on the queue’s stored value. The twelve options outside that dict, dedup_window_seconds and retention_enabled among them, are the ones a partial call now leaves alone.

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 sends traceId only when it is a valid UUID.

The broker’s push path stores no trace id, so it is dropped and the message pops back with traceId: null. Push inside a transaction when a trace id has to survive. See Pick your SDK.

res = await client.queue("orders").partition("customer-42").push([
    {"data": {"orderId": 9137, "amount": 99.5}}
])

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 to a buffered marker once the buffer has accepted the messages, without sending anything. While the buffer sits at max_size the await parks until the flusher drains below the bound, so under overload the producer degrades to the flush rate instead of growing the heap.

Consuming

pop()

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

messages = await client.queue("orders").batch(10).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.

pop_result()

async pop_result() → PopResult(messages, autopilot), a named tuple. It is the same call as pop() and returns the messages plus the broker’s account of how it sized the claim: partitions, batch, and an optional wait_millis pacing hint the consume loop honours in place of its own delay between empty polls. autopilot is None when the pop did not engage autopilot, when the broker is older than 1.2, or when the answer was a bodiless 204, which has no body to carry an echo. See pop autopilot.

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.

The buffer is bounded and lossless under errors. At max_size waiting messages (default 4 * message_count), further adds await the flusher instead of growing the heap. A batch whose POST fails goes back to the front of the buffer, in order, and is retried every retry_delay_millis until it lands or the client closes; it is never dropped. A broker outage therefore shows up as parked producers and a full buffer, not as silent loss.

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

A transaction is atomic, not exactly-once end to end; see the limit.

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 (grace_period, idle_flush_ms, event_time, allowed_lateness, on_late) and are converted before they reach the wire, so the defaults and the close semantics in the JavaScript window option table apply here unchanged. 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, with the proxy’s stable values listed under Behind the proxy. A 204 response yields None: a 204 carries no body at all.

Same surface, other languages

Parity here is a claim about the surface, not about shared code: each client is written natively in its own idiom. These sixteen rows are that surface, read out of clients/ on this tree. Six of them are full parity across all six clients; the other ten are where a client stops.

The ten divergent rows, with their conditions:

• Key/value. The seven operations are get, getMany, getPrefix, put, putIfAbsent, delete and incr. C++ wraps five of them: getMany and getPrefix have no method, and both remain reachable through POST /api/v1/kv on the shared HttpClient. getPrefix inside a transaction is refused by the broker in every client, which is its rule and not a client’s gap.
• Timers. C++ has schedule and cancel only; peek and list are reads and stay on GET /api/v1/timers/{queue}[/{timerKey}]. There is no reschedule operation anywhere, because schedule is the upsert and status reports which happened; Python, Go, Rust and PHP still spell a reschedule alias over it, JavaScript and C++ do not.
• once. JavaScript (kv.once, transaction().once) and Python (kv.once, transaction().once) fold putIfAbsent plus required into the question people actually ask, “did I win?”. The other four write the putIfAbsent themselves, which is the same wire and one line longer.
• kv on a streams operator context. No client has it, so it is not a row. Inside a stream the state primitive is state_ops, which commits with the sink and the ack in the cycle’s own transaction; a key/value write from an operator would not, and that atomicity is the thing the stream already gives you for free.
• Admin facade. C++ has no Admin class. Every management and observability route is still reachable, through the shared HttpClient returned by get_http_client().
• affinity. The C++ LoadBalancer implements round-robin and "session" only, and load_balancing_strategy = "affinity" falls through to round-robin silently. The strategy matters because it keeps one consumer’s pops on one backend, which is what keeps two clients from contending on the same partition claim.
• Streams. The Stream builder, the four window kinds, gates and the /streams/v1/* runtime are in the JavaScript, Python, Go and Rust clients. PHP and C++ have none of it.
• Lease renewal. In C++, renew_lease() sets a flag the consumer loop does not act on: the worker carries an unimplemented placeholder where the timer would go. Call client.renew(message) yourself before the lease expires.
• Per-message trace. Where it exists it is attached by the consume loop, so a message returned by a manual pop() never carries it. Rust records a trace through Admin::record_trace instead of hanging a method off the message. C++ cannot hang one off a JSON object at all, and its trace hook is an explicit no-op.
• Host override. JavaScript (hostHeader) and Rust (host_header) advertise a request authority, and a TLS SNI, independent of the address the socket dials. That is what addresses a named tenant cluster behind a shared proxy endpoint. The other four clients send the dialled address.
• pop failures. Go, Rust and PHP surface a 4xx, an exhausted 429 budget, a terminal 403 or a network fault to the caller, so an empty result means an empty queue. JavaScript, Python and C++ log the failure and return an empty result, which does not distinguish an empty queue from revoked credentials.

The py suite in the test matrix runs the whole pytest tree under clients/client-py/tests, streams included, on the single, ha and tenanted topologies.