Skip to content

datavo examples

Minimal, runnable scripts for the datavo store and dataset surfaces. Each script is self-contained, creates its own throwaway store/dataset, prints what it does, and cleans up after itself.

They are the code companion to the user guide: each script is one chapter's runnable half, in the same order. A store is an append-only operation graph: grown by grow ops (source_import, add) that create sample rows, and enriched by attach ops that write derived keys onto existing samples under a producer. A workset is a live selection over one or more stores that you iterate directly; a dataset is a frozen, named, shard-materialized selection. The prose for all of it starts at Concepts.

Writing — building & operating a store:

Script Guide Use case Key calls
01_ingest_source.py ch. 4 Create an empty store and import a source into it create_store, create_source_import → reserve → upload → commit → wait
02_add_to_store.py ch. 5 Branch a selection of samples into a new store (zero-copy, origin_key lineage) add_to_store → wait_for_add
03_attach_keys.py ch. 6 Enrich a store with a derived key group under a producer attach workset (Not(AttachedBy(...))) → from_workset(..., decode=False) → TarWriter → attach_keys (registers on attach)
04_operations.py ch. 7 Inspect the operation graph and remove a node, gated by dependencies list_store_operations, remove_store_operation (409 when a later attach froze a grow op)

Reading — select a workset and iterate it:

Script Guide Use case Key calls
05_read_samples.py ch. 8 Read a store's live shape, then select a workset over it and iterate its samples directly get_store / list_store_operations vs create_workset → SampleStream.from_workset
06_consume_produced_features.py ch. 9 Consume an attached feature via a producer pin — and compare two producers writing the same key create_workset(..., producer_pins={key: producer}) → SampleStream.from_workset (one workset per producer)
07_union_sources.py ch. 10 Span two stores with one workset — first over the key they share (two sources, no key_groups), then over differently-encoded audio with an any (OR) group, where the consumer coalesces audio.wav or audio.flac create_workset (multiple sources; derived group vs KeyGroupSpec(mode="any")) → SampleStream.from_workset → row.get(...)

Keeping & ML flows — convert to a dataset, then train/eval:

Script Guide Use case Key calls
08_workset_to_dataset.py ch. 11 Keep a selection: watch a live workset follow the store, then convert it to a permanent, reproducible dataset (frozen against later imports) — the bridge to training create_workset → from_workset → convert_workset_to_dataset → SampleStream.from_dataset
09_train_eval_compare.py ch. 12 Convert a workset to a dataset, cut a train/val/test split, score two runs on the same held-out test split, compare their aggregate accuracy build_dataset (create_workset→convert) → create_dataset_split_set → add_to_store(from_dataset="…/test") → get_dataset_scalar_stats
10_nested_pipeline_compare.py ch. 13 A nested pipeline — base model trains on a dataset, follow-up stages train/attach on worksets; model 2 pins & consumes model 1's key, an intermediate is Added into a new store, two decode options are scored against the label base build_dataset + follow-up create_workset (training on a workset) + producer_pins → add_to_store(from_workset=…) → two producers on one decision.cls → pinned from_workset scored vs label

Sharing — who else can reach what you built:

Script Guide Use case Key calls
11_share_a_dataset.py ch. 14 Walk a dataset from private → shared with a team → published to everyone → withdrawn, reading it while published; then preview (never commit) an irreversible ownership transfer get_auth_status (teams) → grant_resource_access / revoke_resource_access (team="everyone" is publishing) → preview_resource_ownership_transfer

Performance — cache shards locally:

Script Guide Use case Key calls
12_warm_and_read_cache.py ch. 15 Read a dataset cold (fetched from the server and cached), then warm through a fresh cache over the same dir (served from local disk, no server); inspect the hierarchy's tiers and the zero-config default SampleStream.from_dataset(..., cache=ShardCacheHierarchy([ShardCacheTier(...)])) → ds.download_stats (downloads vs cache_hits) → get_default_shard_cache()

Running them

Requires datavo-sdk (with the [entra] extra for dev/prod auth) and numpy. Pick a server with DATAVO_SERVER. Everything else (auth, cleanup) is handled by the scripts via _common.py.

Against dev (Entra/EasyAuth — the token is refreshed silently from your MSAL cache; run datavo login once if you have never logged in):

DATAVO_SERVER=https://dev.datavo.io python examples/01_ingest_source.py

Against a local stack (SQLite + simple auth, no token needed — see the "Local development" section of ../CLAUDE.md):

# terminal 1
ALTAVO_DATASET_STORE_AUTH_PROVIDER=simple uv run datavo-api
# terminal 2 — the worker finalizes imports, materializes adds, commits attached
# outputs, and plans datasets, so every example needs it running
uv run datavo-worker --poll-interval-seconds 1
# terminal 3
DATAVO_SERVER=http://localhost:8000 python examples/01_ingest_source.py

If DATAVO_SERVER is unset, the scripts fall back to the active profile in ~/.datavo/config.json (override the profile with DATAVO_PROFILE=<name>).

Notes

  • Server-agnostic auth. _common.make_client() reads /config and picks the right auth automatically: no token for simple, silent MSAL refresh (with an interactive fallback) for entra / easyauth.
  • Reads use a workset; the dataset enters at training. Reading (05–07) selects a workset — a live, directly-iterable selection (create_workset → SampleStream.from_workset); no dataset is needed just to read. A workset stores your selection rather than its result, so creating one costs nothing and every read re-runs it against the store as it is now. When you want a stable, reusable, reproducible source, you convert a workset into a dataset (convert_workset_to_dataset) — introduced in 08 and used for training (09–10, via the _common.build_dataset = create_workset→convert helper).
  • Worker-deferred ops. add_to_store returns immediately and is materialized by a worker; 02_add_to_store.py waits with client.wait_for_add(...). A workset is ready the moment you create it — there is nothing to plan — so wait_for_workset returns on the first poll; converting to a dataset still blocks on wait_for_dataset, since a dataset really is materialized. Run a worker locally — dev has workers already.
  • Idempotency has two mechanisms, and both are used here. Convergence is content-derived and per (producer_id, key): the attach workset comes back empty when nothing is missing, so a re-run is a no-op rather than a recompute — and a half-finished run resumes with exactly the remainder. Replay is exact and per operation: every import, add and attach carries an operation_id you choose, and re-submitting the same one returns the existing operation. Omit it and the SDK mints a fresh one, so nothing replays by accident. These scripts pass stable ids, which is why re-running them is safe. Full rules: idempotency.
  • Datasets pin their store; worksets don't. A dataset holds a retention-tracked dependency on the store(s) it reads, so the dataset examples (08–09) tear the dataset down before the store (_common.cleanup_datasets then cleanup). A workset pins nothing — it holds a selection, not samples — so the read examples (05–07) just drop the store, and the workset is released once nobody is using it.
  • Store vs workset vs dataset. A store is the live, still-growing object you read with get_store / list_store_operations; a workset is a live, unnamed selection over it that you iterate with SampleStream.from_workset; a dataset is a named, frozen, reproducible snapshot you open with SampleStream.from_dataset. Naming a selection is converting it — that is the whole lifetime model.
  • OR (any) key groups. A KeyGroupSpec(mode="any") guarantees at least one of its keys per sample, not a specific one, and datavo does not coalesce them — the consumer reads whichever is present (row.get("audio.wav") or row.get("audio.flac")), as 07_union_sources.py shows.
  • Splitting is a dataset feature. A split set is a seeded, deterministic partition, so it needs a member set that cannot change — which is a dataset, not a live workset. Build the dataset first (create_workset → convert_workset_to_dataset, the _common.build_dataset helper) and cut the split over that; asking for a split over a workset is rejected with a pointer to convert.
  • Split sets are worker-cut child datasets. create_dataset_split_set returns a planning split set; the worker then materializes {split_set}/{train,val,test} as ordinary, ready datasets you can add_to_store(from_dataset=…) or open. 09 waits on it with the wait_for_split_set helper in _common.py (the SDK ships waiters for imports/Adds/datasets but not split sets). Tear a split set down — archive then delete, which cascades its child splits — before its parent dataset and after any store seeded from a split (that Add pins the split child).
  • Scalar keys aggregate server-side. A numeric scalar key (.cls, .count, .index, …) lands as a per-sample attribute, so get_dataset_scalar_stats returns its mean/min/max over a whole dataset without streaming a byte — that is how 08 compares run accuracies (mean(correct.cls)). One caveat drives 09's design: a scalar key is a single attribute per sample (latest writer wins), so two producers writing the same scalar key in one store can't be separated by stats — pin each and read it (the 06 pattern), as 09 does to score its two decode options.
  • Names are examples only. Store names, producer ids, and key names here are illustrative; nothing is invented against a real environment.