Queues & Dicts

Overview

bladekit.Queue and bladekit.Dict are distributed primitives backed by Postgres on the control plane. They let independent functions — or a local script and a remote worker — exchange data without wiring up your own storage: a queue for producer/consumer work distribution, a dict for shared key-value state (config broadcast, progress tracking, dedup sets, checkpoints).

Both are identified by name and created lazily: from_name does not make a round trip, the store is get-or-created by the server on the first put/get. Two from_name calls with the same name — from any function, container, or process — address the same underlying store.

bladekit.Queue

A distributed FIFO queue. put appends to the tail; get atomically claims and removes the head item, so two concurrent consumers never receive the same item.

class Queue:
    @classmethod
    def from_name(cls, name: str, *, create_if_missing: bool = False) -> "Queue": ...

    @property
    def name(self) -> str: ...

name str

Queue name — global namespace. Identifies the store; reused across calls to reference the same queue.

create_if_missing boolDefault: False

Accepted for parity with Volume.from_name (and Modal). It is a no-op — queues are always get-or-create on first use regardless of this value.

Queue.put / Queue.get

def put(self, value: Any) -> None: ...
def get(self, *, timeout: int = 0) -> Optional[Any]: ...

value Any

Item to append to the tail of the queue. Serialized with cloudpickle.

timeout intDefault: 0

With timeout=0, get() returns immediately — None if the queue is empty. With timeout > 0, the call long-polls on the server for up to timeout seconds and returns None if it times out with no item.

Queue.len / Queue.delete

def len(self) -> int: ...
def delete(self) -> None: ...

len() returns the current item count. It is non-transactional and best-effort — another consumer may claim items between the request and the value you observe, so treat it as an approximation, not a precise snapshot for coordination logic.

delete() removes the queue and all of its items.

bladekit.Dict

A distributed key-value store with the same get-or-create-by-name model as Queue. put is an upsert — last-write-wins per key, with no transaction across keys.

class Dict:
    @classmethod
    def from_name(cls, name: str, *, create_if_missing: bool = False) -> "Dict": ...

    @property
    def name(self) -> str: ...

name str

Dict name — global namespace. Reused across calls to reference the same store.

create_if_missing boolDefault: False

Accepted for parity with Volume.from_name; a no-op — the store is always get-or-create on first use.

Dict.get / put / contains / delete

def get(self, key: str, default: Any = None) -> Any: ...
def put(self, key: str, value: Any) -> None: ...
def contains(self, key: str) -> bool: ...
def delete(self, key: str) -> None: ...

key str

UTF-8 string key, ≤ 256 bytes encoded. get/put/contains/delete all validate this and raise InvalidArgumentError (DICT_KEY_TOO_LARGE) if exceeded, or if key is not a str.

value Any

Value to store, serialized with cloudpickle. ≤ 1 MiB inline; larger values spill to the object store automatically.

default AnyDefault: None

Value returned by get when key is absent.

contains(key) checks existence without fetching or deserializing the value — prefer it over get(key) is not None when the value itself may legitimately be None.

Dict[] accessors

Dict implements the mapping-style accessors, layered on get/put/contains/delete:

Example: producer / consumer

A producer function enqueues work items and a consumer function drains them, reporting progress through a shared Dict:

import bladekit

app = bladekit.app("queue-dict-demo")
image = bladekit.Image.from_catalog("fabric/base")

@app.function(image=image)
def produce(urls: list[str]):
    q = bladekit.Queue.from_name("jobs")
    progress = bladekit.Dict.from_name("job-progress")
    for url in urls:
        q.put({"url": url})
    progress["total"] = len(urls)
    progress["done"] = 0

@app.function(image=image, gpu="RTX-A4500")
def consume():
    q = bladekit.Queue.from_name("jobs")
    progress = bladekit.Dict.from_name("job-progress")
    while True:
        item = q.get(timeout=30)
        if item is None:
            break  # no work for 30s — assume the queue is drained
        process(item["url"])
        progress["done"] = progress.get("done", 0) + 1

@app.local_entrypoint()
def main():
    produce.remote(["https://example.com/a", "https://example.com/b"])
    consume.remote()

Related reference