Queue

Queues: durable background work, delivered at least once.

jobs = noclick.Queue.from_name("invoices", create_if_missing=True)
jobs.put({"invoice": "INV-1042"})

for msg in jobs.iterate(lease_s=60):
    print("Sending", msg.body["invoice"])
    msg.ack()

A message you take is leased to you for lease_s seconds: ack() it when the work is done, or it comes back for another try (so handle a message twice safely). A message that fails max_attempts times moves to the dead letters, where you can retry or discard it. An agent can take the messages instead: triggers=[noclick.OnQueue(jobs)] makes each one a turn.

class Queue

A durable queue of JSON messages, delivered at least once.

put adds work; get and iterate take it under a lease, and each message is acked when done or delivered again. depth is how many wait, in_flight how many are leased, dead_letter_count how many failed max_attempts times (as of the last load; len() reads the depth fresh). used_by lists who consumes it: agents with an OnQueue trigger and apps that bind it.

Like every live object, it also has hydrate(), refresh() and raw.

Attributes

id	str
name	str
depth	int	Messages waiting to be taken, delayed ones included.
in_flight	int	Messages leased and not yet acked.
dead_letter_count	int	Messages that failed `max_attempts` times.
max_attempts	int	Deliveries a message gets before it's a dead letter.
used_by	list[dict[str, Any]]	Who consumes it: agents with an `OnQueue` trigger (`kind` `agent`) and apps that bind it (`kind` `project`, with the `binding` name).
region	str \| None	The region it keeps its data in (`noclick.regions()`), fixed when it was created.
user_id	str \| None
created_at	Any
console_url	str \| None	This queue in the developer console, with its messages and dead letters (None off the hosted API).

createclassmethodasync

noclick.Queue.create(
    name, *, max_attempts=5, region=None, user=..., client=None
) -> Queue

Create a queue called name (1-64 letters, digits and . _ -), unique to its user (a taken name raises noclick.Conflict with existing_id). A message is delivered up to max_attempts times (1 to 100) before it's a dead letter. region (noclick.regions(kind="queue")) is where its messages are kept, for good; without one, your account's default region, else us-east. Queues are kept in us-east only for now: another region, or a default region elsewhere, raises noclick.InvalidRequest naming it.

Parameters

namestrrequired

max_attemptsintdefault 5

regionstr | Nonedefault None

userAny

Who the call acts for. Defaults to the user set by noclick.as_user (none: the account itself).

(defaults to the current user)

clientClient | None

The client to send with; defaults to the module client built from NOCLICK_API_KEY.

(default None)

Returns

Queue

The new queue, empty.

Usage

A queue of the account's own:

jobs = noclick.Queue.create("invoices")

Give flaky work more tries, for one user:

with noclick.as_user("acme"):
    exports = noclick.Queue.create("erp-exports", max_attempts=10)

from_nameclassmethod

noclick.Queue.from_name(
    name, *, create_if_missing=False, max_attempts=None, region=None, user=...,
    client=None
) -> Queue

A lazy reference to the queue called name in the current (or given) user. With create_if_missing=True it's created on first use, in region (as create), with max_attempts (5 if you leave it out); a max_attempts you pass is also applied to a queue that already exists, so your code stays the definition.

Inside a deployed app (signed in by its deployment) the name is looked up in the app's own namespace for the environment the code runs in, and one made there is the app's: a name bound to the app by hand wins, and in a preview of an app whose preview_resources is branch the branch gets its own empty queue, so a preview never changes production's data.

Parameters

namestrrequired

create_if_missingbooldefault False

max_attemptsint | Nonedefault None

regionstr | Nonedefault None

userAny

Who the call acts for. Defaults to the user set by noclick.as_user (none: the account itself).

(defaults to the current user)

clientClient | None

The client to send with; defaults to the module client built from NOCLICK_API_KEY.

(default None)

Returns

Queue

A queue that loads (or is created) on first use. Nothing is sent yet.

Usage

Make your code the definition, so the first run creates it:

jobs = noclick.Queue.from_name("invoices", create_if_missing=True)
jobs.put({"invoice": "INV-1042"})

Change how many tries a message gets:

buffer = noclick.Queue.from_name("webhook-buffer", create_if_missing=True, max_attempts=8)
print(buffer.max_attempts)   # 8

from_idclassmethod

noclick.Queue.from_id(queue_id, *, client=None, user=...) -> Queue

A lazy reference to a queue by id. There's no Queue.get(id), since get takes a message: call hydrate() on this to load it now.

Parameters

queue_idstrrequired

clientClient | None

The client to send with; defaults to the module client built from NOCLICK_API_KEY.

(default None)

userAny

Who the call acts for. Defaults to the user set by noclick.as_user (none: the account itself).

(defaults to the current user)

Returns

Queue

A queue that loads the first time you use it. Nothing is sent yet.

Usage

jobs = noclick.Queue.from_id("8d3f1a6c-2b7e-4c95-9e10-4a6b2c8d7f31").hydrate()
print(jobs.name, jobs.depth, jobs.dead_letter_count)

listclassmethodasync

noclick.Queue.list(
    *, limit=None, user_filter=None, user=..., client=None
) -> list[Queue]

The current user's queues (the account's own outside as_user, or those user_filter covers).

Parameters

limitint | None

Return at most this many.

(default None)

user_filterAny

Which users a read covers: noclick.ALL_USERS, noclick.NO_USER, a User, or your id for one. It filters without acting as that user.

(default None)

userAny

Who the call acts for. Defaults to the user set by noclick.as_user (none: the account itself).

(defaults to the current user)

clientClient | None

The client to send with; defaults to the module client built from NOCLICK_API_KEY.

(default None)

Returns

list[Queue]

Every matching queue, newest first, fetched page by page.

Usage

How deep each of your queues is:

for queue in noclick.Queue.list():
    print(queue.name, queue.depth, queue.in_flight, queue.dead_letter_count)

Every user's:

every = noclick.Queue.list(user_filter=noclick.ALL_USERS)

list_pageclassmethodasync

noclick.Queue.list_page(
    *, cursor=None, limit=None, user_filter=None, user=..., client=None
) -> Page

One page of queues; pass next_cursor back as cursor.

Parameters

cursorstr | None

The next_cursor of the page before; omit it for the first page.

(default None)

limitint | None

Return at most this many.

(default None)

user_filterAny

Which users a read covers: noclick.ALL_USERS, noclick.NO_USER, a User, or your id for one. It filters without acting as that user.

(default None)

userAny

Who the call acts for. Defaults to the user set by noclick.as_user (none: the account itself).

(defaults to the current user)

clientClient | None

The client to send with; defaults to the module client built from NOCLICK_API_KEY.

(default None)

Returns

Page

A page whose data holds the queues and whose next_cursor is None on the last page.

Usage

page = noclick.Queue.list_page(limit=50, user_filter=noclick.ALL_USERS)
for queue in page.data:
    print(queue.user_id, queue.name, queue.depth)

update async

queue.update(*, max_attempts) -> Queue

Change how many deliveries a message gets before it's a dead letter (1 to 100). Messages already on the queue follow the new number.

Parameters

max_attemptsintrequired

Returns

Queue

This queue, with its new max_attempts.

Usage

jobs = noclick.Queue.from_name("invoices")
jobs.update(max_attempts=10)

put async

queue.put(body, *, delay_s=0, dedupe_key=None) -> str

Add a message: any JSON, up to 256 KB. With delay_s (up to 7 days) nobody can take it until then, for a retry later or a reminder. dedupe_key (up to 128 characters) makes the put safe to repeat: for 24 hours, another put with the same key adds nothing and returns the first message's id, so code that crashes and puts again doesn't enqueue twice.

Parameters

bodyAnyrequired

delay_sintdefault 0

dedupe_keystr | Nonedefault None

Returns

str

The new message's id (the first one's, for a repeated dedupe_key).

Usage

Hand a job to your workers:

jobs = noclick.Queue.from_name("invoices", create_if_missing=True)
jobs.put({"invoice": "INV-1042", "customer": "acme"})

Check back on a payment in an hour:

jobs = noclick.Queue.from_name("invoices", create_if_missing=True)
message_id = jobs.put({"check_payment": "INV-1042"}, delay_s=3600)
print(message_id)

Enqueue each order once, however often this code runs for it:

jobs = noclick.Queue.from_name("fulfilment", create_if_missing=True)
jobs.put({"order": "o_981"}, dedupe_key="order-o_981")

put_many async

queue.put_many(bodies, *, delay_s=0) -> list[str]

Add up to 100 messages in one call, all of them or none.

Parameters

bodiesSequence[Any]required

delay_sintdefault 0

Returns

list[str]

The new messages' ids, in the order you gave them.

Usage

jobs = noclick.Queue.from_name("invoices", create_if_missing=True)
ids = jobs.put_many([{"invoice": f"INV-{n}"} for n in range(1042, 1052)])
print(len(ids))   # 10

get async

queue.get(*, timeout_s=0, lease_s=60) -> QueueMessage | None

Take the next message, waiting up to timeout_s seconds (at most 60) for one to arrive. It's leased to you for lease_s seconds (up to 12 hours): ack() it before then, or it's delivered again. Messages come out roughly in the order they went in, but order isn't guaranteed.

Parameters

timeout_sintdefault 0

lease_sintdefault 60

Returns

QueueMessage | None

The message, or None if none arrived in time.

Usage

jobs = noclick.Queue.from_name("invoices")
msg = jobs.get(timeout_s=30, lease_s=120)
if msg is not None:
    print(msg.body, msg.attempts)
    msg.ack()

iterate iterator

queue.iterate(*, lease_s=60, stop_when_empty=False) -> Iterator[QueueMessage]

Take messages as they arrive, each leased for lease_s seconds; it waits while the queue is empty and runs until you stop iterating. With stop_when_empty=True it ends instead the first time nothing is waiting (messages leased to someone else, or delayed, don't keep it going), to drain a queue in a batch job. Ack each message when its work is done. async for msg in queue.iterate.aio() in async code.

Parameters

lease_sintdefault 60

stop_when_emptybooldefault False

Returns

Iterator[QueueMessage]

Each message as it's taken. Without stop_when_empty it never ends on its own, so break when you're done.

Usage

A worker:

jobs = noclick.Queue.from_name("invoices", create_if_missing=True)
for msg in jobs.iterate(lease_s=120):
    try:
        print("Sending", msg.body["invoice"])
    except Exception:
        msg.nack(delay_s=30)
    else:
        msg.ack()

In async code:

jobs = noclick.Queue.from_name("invoices", create_if_missing=True)
async for msg in jobs.iterate.aio(lease_s=120):
    print(msg.body)
    await msg.ack.aio()

Drain what's there, then finish:

jobs = noclick.Queue.from_name("invoices")
for msg in jobs.iterate(stop_when_empty=True):
    print("Sending", msg.body["invoice"])
    msg.ack()

len async

queue.len() -> int

How many messages wait (delayed ones included, leased ones and dead letters not), read now.

Returns

int

The queue's depth.

Usage

jobs = noclick.Queue.from_name("invoices")
print(jobs.len(), "invoices waiting")

peek async

queue.peek(*, limit=10) -> list[QueueMessage]

Look at the oldest messages without taking them: waiting and in flight, oldest first. Nothing is leased, so you can't ack these.

Parameters

limitint

Return at most this many.

(default 10)

Returns

list[QueueMessage]

Up to limit (at most 100) messages, with their status and attempts.

Usage

jobs = noclick.Queue.from_name("invoices")
for msg in jobs.peek(limit=5):
    print(msg.status, msg.attempts, msg.body)

dead_letters async

queue.dead_letters(*, limit=None) -> list[QueueMessage]

The messages that failed max_attempts times, newest first. Each says why in last_error; retry() or discard() it. Dead letters are kept 14 days.

Parameters

limitint | None

Return at most this many.

(default None)

Returns

list[QueueMessage]

Every dead letter (or the first limit), fetched page by page.

Usage

jobs = noclick.Queue.from_name("invoices")
for letter in jobs.dead_letters():
    print(letter.body, letter.attempts, letter.last_error)

depth_history async

queue.depth_history(*, hours=24) -> list[QueueDepth]

How deep the queue has been over the last hours (up to 168), a point every 15 minutes: whether your workers keep up.

Parameters

hoursintdefault 24

Returns

list[QueueDepth]

A QueueDepth (at, depth, in_flight) per point, oldest first.

Usage

jobs = noclick.Queue.from_name("invoices")
worst = max(jobs.depth_history(hours=24), key=lambda point: point.depth)
print(worst.at, worst.depth)

clear async

queue.clear() -> None

Drop every waiting message and every dead letter. Messages in flight finish as usual. The queue itself stays.

Usage

jobs = noclick.Queue.from_name("invoices")
jobs.clear()

export async

queue.export() -> Operation

Export every message as it stands at one moment, dead letters included, as JSON Lines: one per line, oldest first, with its status (waiting, delayed, in_flight or dead), attempts and body.

Returns

Operation

An operation whose wait() gives an ExportBundle: a zip at url holding the files.

Usage

jobs = noclick.Queue.from_name("invoices")
bundle = jobs.export().wait()
print(bundle.url)

delete async

queue.delete() -> TrashItem

Delete the queue and every message on it: it goes to the trash, restorable for 30 days. A queue an app binds or an agent's OnQueue trigger reads raises noclick.Conflict until you unbind it or remove the trigger.

Returns

TrashItem

Its TrashItem: restore() puts it back, messages and all.

Usage

jobs = noclick.Queue.from_name("invoices")
jobs.delete()

QueueMessage

class QueueMessage

One message on a queue. body is the JSON you put; attempts counts its deliveries, this one included. A message you took with get or iterate is leased to you until lease_expires_at: ack() when the work is done, nack() to hand it back now as a failed try, release() to hand it back without spending one. A dead letter (from dead_letters()) says why its last attempt failed in last_error; retry() puts it back on the queue and discard() deletes it.

status is waiting (delayed until visible_at), in_flight while leased, or dead.

Like every live object, it also has hydrate(), refresh() and raw.

Attributes

id	str
queue_id	str
body	Any	The message as you put it: JSON.
status	str	`waiting`, `delayed`, `in_flight` or `dead`.
attempts	int	How many times it has been delivered, this delivery included.
enqueued_at	Any
visible_at	Any	When a delayed message can first be taken (None once it can).
lease_expires_at	Any	When your lease ends and the message goes back (None unless you hold it).
last_error	str \| None	Why its last attempt failed (None if none has).
dead_lettered_at	Any	When it became a dead letter (None unless it is one).

ack async

msg.ack() -> None

Mark the message done: it's deleted and never delivered again. Ack only after the work has succeeded. Once the lease has run out the message may already be someone else's, so a late ack raises noclick.Conflict; your work must then be safe to do twice.

Usage

jobs = noclick.Queue.from_name("invoices")
msg = jobs.get(timeout_s=30)
if msg is not None:
    print("Sending", msg.body["invoice"])
    msg.ack()

nack async

msg.nack(*, delay_s=0, error=None) -> None

Hand the message back unfinished, for another try after delay_s seconds. It counts as an attempt: after the queue's max_attempts it moves to the dead letters instead. error (up to 2 KB) says why; it's the message's last_error, which a dead letter keeps.

Parameters

delay_sintdefault 0

errorstr | Nonedefault None

Usage

jobs = noclick.Queue.from_name("invoices")
msg = jobs.get(timeout_s=30)
if msg is not None:
    msg.nack(delay_s=60, error="The invoicing service answered 503")   # try again in a minute

release async

msg.release(*, delay_s=0) -> None

Hand the message back untouched, for someone to take after delay_s seconds, without spending an attempt: attempts goes back to what it was before this delivery, and last_error is left as it was. Use it when the failure isn't the message's (your account is out of money, a service everything depends on is down), so a backlog doesn't burn its tries and land in the dead letters. A message you release keeps coming back until the cause is fixed, so stop taking more meanwhile.

Parameters

delay_sintdefault 0

Usage

jobs = noclick.Queue.from_name("recordings")
msg = jobs.get(timeout_s=30)
if msg is not None:
    try:
        words = noclick.AI.transcribe(msg.body["url"])
    except noclick.PaymentRequired:
        msg.release(delay_s=3600)   # not the recording's fault: no try spent
    else:
        print(words.text)
        msg.ack()

retry async

msg.retry() -> None

Put a dead letter back on the queue, with its attempts reset, once you've fixed what made it fail.

Usage

jobs = noclick.Queue.from_name("invoices")
for letter in jobs.dead_letters():
    letter.retry()

discard async

msg.discard() -> None

Delete a dead letter for good.

Usage

jobs = noclick.Queue.from_name("invoices")
for letter in jobs.dead_letters():
    if "unknown invoice" in (letter.last_error or ""):
        letter.discard()

QueueDepth

class QueueDepth

How full a queue was at one moment: depth messages waiting and in_flight leased, at at.

Attributes

at	Any
depth	int
in_flight	int