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), then find the source import's samples in it	add_to_store → wait_for_add → SourceSpec(filters=[ImportedBy(operation_id)])
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; last, combine two existing worksets with sources_of, each part keeping its own keys	create_workset (multiple sources; derived group vs KeyGroupSpec(mode="any")) → SampleStream.from_workset → row.get(...) → sources_of(...)

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_split_a_dataset.py	ch. 12	Cut a dataset into train/val/test with the split engine: discover split fields, watch ratios leak a speaker then fix it with an exclusivity rule (+ stratify a locale), verify determinism, and carve a split by attribute value	get_dataset_split_fields → preview_dataset_split_set → create_dataset_split_set (ExclusivityRule / StratificationRule / SplitAttributeSelector) → wait_for_split_set
10_train_eval_compare.py	ch. 13	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
11_nested_pipeline_compare.py	ch. 14	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

Access — who else can reach what you built, and how a program acts for you:

Script	Guide	Use case	Key calls
12_share_a_dataset.py	ch. 15	Walk a dataset from private → shared with a team → published to everyone → withdrawn, reading it while published; then hand a program a run token that reads the dataset as you but is refused when it tries to share it, and revoke the token	get_auth_status (teams) → grant_resource_access / revoke_resource_access (team="everyone" is publishing) → mint_run_token → DatavoClient(auth_token=…) → revoke_run_token

Performance — cache shards locally:

Script	Guide	Use case	Key calls
13_warm_and_read_cache.py	ch. 16	Warm a dataset cold with warm_cache (fetched from the server into the cache without decoding a sample), then read it through a fresh cache over the same dir (served from local disk, no server); inspect the hierarchy's tiers and the zero-config default	warm_cache(SampleStream.from_dataset(..., cache=ShardCacheHierarchy([ShardCacheTier(...)]))) → ds.download_stats (downloads vs cache_hits) → get_default_shard_cache()
14_manual_split.py	ch. 17	Materialize a partition you already have: hand-pick each split's ids (with a stale id and some samples left uncovered), upload the membership multi-part tagged by split, create an id_list split set, and read the drop/overlap diagnostics	upload_split_membership → create_dataset_split_set (SplitIdListSelector) → wait_for_split_set → detail.manual
15_member_of.py	ch. 18	Reuse a split's membership: read the holdout with a key the dataset never froze, read it in a store its samples were added into (and see membership_matches), then read everything except the holdout; last, union the splits back with sources_of, and see an overlapping union hold each sample once, live and frozen alike	SourceSpec(filters=[MemberOf(dataset, split_set=…, split_name=…)]) → get_workset(...).membership_matches → Not(MemberOf(...)) → sources_of(split)

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 (10–11, 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–11) 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 and 10 wait 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 10 compares run accuracies (mean(correct.cls)). One caveat drives 11'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 11 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.