Skip to content

4. Ingest a source

Runnable companion: examples/01_ingest_source.py

A source import is how raw data becomes samples. It is the boundary between the outside world and Datavo: you hand it tar archives with whatever ids your source system uses, and it hands back canonical samples with server-assigned sample_ids.

Create the store first

Stores are explicit. Nothing creates one as a side effect, so a typo in a store name is an error rather than a new namespace:

client.create_store(
    name="voice_corpus_v1",
    display_name="Voice corpus v1",
    metadata={"owner": "speech-team"},      # free-form, yours
)

The tar shape

Datavo ingests WebDataset-style tars. There is one naming rule: each member is <sample-key>.<key>, split at the first dot.

0001.audio.wav        →  sample "0001", key "audio.wav"
0001.transcript.txt   →  sample "0001", key "transcript.txt"
0002.audio.wav        →  sample "0002", key "audio.wav"

Every distinct prefix becomes one sample, carrying whatever keys appear under it. Nothing else is required: samples in one tar need not carry the same keys, and no key is privileged. Prefixes need to be unique only within the tar — they are not sample identity, so they need no global scheme; Datavo mints the canonical sample_id and keeps the mapping for provenance.

The SDK's TarWriter writes that layout from dicts:

from datavo_sdk import TarWriter

with TarWriter("batch_001.tar") as writer:
    writer.write({"__key__": "0001", "audio.wav": audio_bytes, "transcript.txt": "hello"})
    writer.write({"__key__": "0002", "audio.wav": other_bytes, "transcript.txt": "world"})

A __meta__.json header is the WebDataset convention and is optional here — the import skips it. Pass tar_meta={...} to TarWriter when something downstream of you reads it.

Import it

Four calls: open the import, upload parts, commit, wait.

imp = client.create_source_import(
    sample_store="voice_corpus_v1",
    operation_id="demo-batch-001",      # yours; running it again replays
    default_tags={"language": "en"},    # applied to every sample it introduces
)

client.upload_source_import_part(imp.source_import_id, 0, "batch_001.tar")
client.commit_source_import(imp.source_import_id)

done = client.wait_for_source_import(imp.source_import_id, timeout_seconds=180)
print(done.state, done.sample_count)     # "committed", 2

upload_source_import_part bundles reserve → upload → complete for one part, and takes a path, bytes or a file object. Upload as many parts as you like before committing — one import can carry a whole session's worth of tars, and parts upload in parallel if you want them to.

Commit is worker-deferred: it queues the work, and a worker reads the tars, stages what it found, and publishes the samples. wait_for_source_import polls to a terminal state and returns the server's authoritative count. Nothing is visible in the store until that finishes.

Running it twice

Two runs of the same importer should not produce two copies. The mechanism is replay, and its key is operation_id.

operation_id is yours — a studio session id, an export job id, whatever names this batch in your vocabulary. Datavo compares it by exact string equality and never derives it. Calling create_source_import again with the same one returns the existing import instead of opening a second, so a retried job resumes rather than duplicating.

Omit it and the SDK generates a fresh id for you, so an interactive import opens a new import every time. That is the intended default: replay is something you opt into by passing a stable id, never something that surprises you.

The id is also how you find the import again:

client.get_source_import_by_operation_id("voice_corpus_v1", "demo-batch-001")
client.list_source_import_samples_by_operation_id("voice_corpus_v1", "demo-batch-001")

And it labels the import in the store's operation log (chapter 7), so "where did this row come from" answers in your words rather than an opaque id.

A failed import releases its id. If an import ends failed, submitting the same operation_id again starts a new one rather than returning the failure — otherwise a stable id would be unusable after its first bad day.

The archive hash rejects; it does not replay

Datavo also hashes the committed parts, and re-committing byte-identical parts into the same store fails — source import archive already committed as <id> — whatever operation_id you used.

This is a different thing from replay and it is worth keeping straight:

  • Same operation_id → replay → you get the first import back, no error.
  • Same bytes, new operation_id → rejection → a second import opens and then fails, and wait_for_source_import raises.

Look an archive up with get_source_import_by_archive(store, "<sha256>"). The full rules for both are in idempotency.

Tags

default_tags are applied to every sample the import introduces, and they are how a workset later carves the catalog ("everything English"). Beyond them, any key whose payload decodes to a scalar — .cls, .count, a numeric or short text value — automatically becomes a searchable attribute on the sample. Those two are the only sources of sample attributes; nothing writes a tag directly.

What you have now

detail = client.get_store("voice_corpus_v1")
detail.sample_count                       # 2
[k.key for k in detail.keys]              # ["audio.wav", "transcript.txt"]

Chapter 8 iterates those samples. The next two chapters finish the write side: bringing samples in from elsewhere, and adding derived keys.

Pitfalls

  • Treating the commit response as final. It reports that the work was queued. state and sample_count are only authoritative after the wait.
  • Reusing an operation_id for different bytes. You get the first import back, not a new one — that is replay doing its job. Use a fresh operation_id per batch of raw data.
  • Re-uploading identical bytes under a new operation_id. The archive hash rejects it and the import fails; import the batch once, or change the data.
  • A dot in the sample key. The first dot is the split, so take.1.audio.wav imports as sample take with key 1.audio.wav. It fails silently — keep prefixes dot-free.

Next: 5. Add samples to a store — grow a store from data that is already in Datavo.