Ingest jobs#

An ingest job tracks a batch of files as one unit, with one status and one entry on the Ingestion page. You get one from IngestBuilder or from a file extractor.

To upload a single file, use the dataset’s own methods, as in Datasets.

Connect the client#

This guide assumes you have installed the client and setup your API keys.

from nominal.core import NominalClient

client = NominalClient.from_profile("default")

Submit a batch#

IngestBuilder uploads several files in parallel, then starts one job for them. It’s experimental, so the API may change between releases.

from datetime import datetime, timezone

from nominal.experimental.ingest import IngestBuilder

dataset = client.get_dataset("<DATASET_RID>")
job = (
    IngestBuilder(client, dataset)
    .add_tabular_data(
        "flight_data.csv",
        timestamp_column="source_time",
        timestamp_type="iso_8601",
        units={"altitude": "m", "airspeed": "m/s"},
    )
    .add_mcap("recording.mcap", exclude_topics=["/diagnostics"])
    .add_video("nose_camera.mp4", "nose_camera", start=datetime(2026, 9, 10, 14, 38, 19, tzinfo=timezone.utc))
    .add_tags({"test_event": "ft-114"})  # applied to every file
    .submit()
)

The add_* methods match the dataset upload methods. add_tabular_data also takes units, channel_prefix and channel_name_overrides, to rename vendor columns as they ingest.

submit returns once the job starts, not when its files finish. A builder submits once. Make a new one for the next batch.

  • If an upload fails, submit raises NominalIngestUploadFailed and starts no job. submit(allow_partial=True) starts the job with the files that uploaded and logs the rest.

  • submit(runs_to_expand=["<RUN_RID>"]) widens those runs to cover the new data.

  • Files in one job share a single nominal_ingest tag. To tell them apart later, pass your own tags to each add_* call.

If submit raises after sending its request, the job may still have started. Search for it before you retry, so you don’t ingest the files twice.

Wait for the files#

as_files_ingested() yields each file as it finishes, including files the job produces while you wait:

from datetime import timedelta

from nominal.core.dataset_file import IngestStatus

for file in job.as_files_ingested(timeout=timedelta(minutes=30)):
    if file.ingest_status is IngestStatus.SUCCESS:
        print(file.name, "ingested")
    else:
        print(file.name, file.ingest_status.name, file.get_ingest_error())

  • A completed job can still have failed files, so check each file’s status.

  • A failed job raises NominalIngestFailed, after yielding the files that finished.

  • A timeout raises NominalIngestTimeout but doesn’t stop the job. cancel() does, without removing data that already ingested.

To wait on files you already have, use wait_for_files_to_ingest(). It returns (done, not_done) and doesn’t see files the job adds later.

Find jobs#

search_ingestion_jobs() filters by dataset, creator, status, text and start time:

from datetime import datetime, timedelta, timezone

from nominal.core import IngestionJobStatus

failed = client.search_ingestion_jobs(
    statuses=[IngestionJobStatus.FAILED],
    start_time_after=datetime.now(timezone.utc) - timedelta(days=1),
)
for job in failed:
    print(job.ingest_type, job.origin_files, job.nominal_url)

A job’s status doesn’t change until you call job.refresh().