Skip to content

Persisting Documentation

Queue

DeepLink-org/Persisting

Queue API¶

`Queue`¶

High-level persistent queue backed by Lance storage.

from persisting import Queue

q = Queue(
    name: str,
    storage_path: str = "./data",
    *,
    batch_size: int = 100,
    auto_flush_interval_sec: float = 0.0,
    enable_metrics: bool = False,
    num_buckets: int = 4,
    bucket_column: str = "id",
    zerocopy_mode: str = "auto",
)

Parameter	Description
`name`	Topic name — used as subdirectory under `storage_path`
`storage_path`	Root directory for queue data
`batch_size`	Auto-flush when buffer reaches this size
`auto_flush_interval_sec`	Periodic flush (0 = disabled)
`enable_metrics`	Collect put/get/flush counters
`num_buckets`	Hash bucket count (distributed mode)
`bucket_column`	Column for consistent hashing
`zerocopy_mode`	`"auto"`, `"force"`, or `"off"`

Write Methods¶

await q.put(record: dict) → BatchMeta | None
await q.put(data: Any, partition_id: str = "default") → BatchMeta | None
await q.put_batch(records: list[dict], partition_id: str = "default") → None
await q.flush() → None

put() accepts either plain dicts (local mode) or tensor data via partition_id (distributed).

Read Methods¶

await q.get(limit: int = 100, offset: int = 0) → list[dict]
await q.get_meta(fields, batch_size, task_name, partition_id, sampler) → BatchMeta
await q.get_data(batch_meta: BatchMeta, partition_id) → Any
await q.get_batch(fields, batch_size, task_name, partition_id, sampler) → Any
await q.stream(limit, offset, wait=False, timeout=None) → AsyncIterator[list[dict]]

Consumption Tracking¶

await q.mark_consumed(task_name: str, global_indexes: list[int], partition_id: str = "default") → None
await q.reset_consumption(task_name: str, partition_id: str = "default") → None
await q.clear(global_indexes: list[int], partition_id: str = "default") → None

Stats¶

await q.stats() → dict
len(q) → int
q.close() → None

`KVInterface`¶

Key-value access layer on top of Queue.

from persisting import Queue, KVInterface

q = Queue("kv_store", storage_path="./data")
kv = KVInterface(q)

Method	Description
`await kv.kv_put(key, data, partition_id, tag=None)`	Put by key
`await kv.kv_batch_put(keys, data, partition_id, tags=None)`	Batch put
`await kv.kv_batch_get(keys, partition_id, fields=None)`	Get by keys
`await kv.kv_list(partition_id)`	List key→tag pairs
`await kv.kv_clear(keys, partition_id)`	Remove keys

Samplers¶

from persisting import SequentialSampler, RankAwareSampler, GRPOGroupNSampler

Sampler	Description
`SequentialSampler()`	Simple sequential sampling
`RankAwareSampler()`	Each rank gets independent slices
`GRPOGroupNSampler(n_samples_per_prompt)`	N samples per prompt group

Internal Backends¶

These are used internally by Queue. Most users should not use them directly.

`LanceBackend`¶

from persisting.queue import LanceBackend

backend = LanceBackend(bucket_id, storage_path, batch_size=100, **kwargs)
await backend.put(record)
await backend.get(limit, offset)
await backend.flush()
await backend.stats()

`PersistingBackend`¶

Extends LanceBackend with metrics:

from persisting.queue import PersistingBackend

backend = PersistingBackend(bucket_id, storage_path, batch_size=100, enable_metrics=True)
await backend.stats()           # includes "metrics" key
await backend.get_metrics()     # metrics snapshot