Write an extractor in Python#

The container contract is environment-driven: input paths and parameters arrive as environment variables, and output goes to the directory named by $OUTPUT_DIR. You can implement it in any language. In Python, the nominal.experimental.extractor package in the nominal client connects your parser to that contract. Declare the files and settings your function needs, receive them as Python arguments, and use ctx to describe the outputs. Those same declarations supply the metadata needed to register the image.

Warning

The authoring helpers in nominal.experimental.extractor are experimental. The container contract they implement is stable, but the Python API may change between releases.

Why the decorators look like Click#

A command-line tool and an extractor face the same setup problem: external inputs need to become usable Python values before the actual work starts. Click expresses that setup with an argument for a positional input and an option for a named setting:

from pathlib import Path
import click

@click.command()
@click.argument("source", type=click.Path(exists=True, dir_okay=False, path_type=Path))
@click.option("--prefix", default="copied", help="Output file name without the extension.")
def copy_csv(source: Path, prefix: str) -> None:
    Path(f"{prefix}.csv").write_bytes(source.read_bytes())

if __name__ == "__main__":
    copy_csv()

Running python cli.py source.csv --prefix flight gives the callback a Path and the string "flight". Click handles lookup and defaults before invoking it. The function can concentrate on copying the data.

An extractor receives mounted files and environment variables instead of command-line tokens. @input and @parameter describe that boundary in the same way. The framework handles resolution, conversion, and validation; your callback handles the data. Click is the inspiration for this API, not a dependency needed to run an extractor.

Start with a declarative extractor#

This CSV example needs only the standard library and the nominal package. Save it as copy_csv.py:

from pathlib import Path
from nominal.experimental import extractor

@extractor.manifest_extractor(
    default_timestamp_column="time",
    default_timestamp_type="epoch_seconds",
)
@extractor.input("source", file_suffixes=["csv"])
@extractor.parameter("prefix", default="copied", description="Output file name without the extension")
def copy_csv(ctx: extractor.ManifestExtractorContext, *, source: Path, prefix: str) -> None:
    output = ctx.output_dir / f"{prefix}.csv"
    output.write_bytes(source.read_bytes())
    ctx.add_tabular(output, timestamp_column="time", timestamp_type="epoch_seconds")

if __name__ == "__main__":
    copy_csv.run()

Try it with a small input:

mkdir -p /tmp/csv-output
printf 'time,value\n1,42\n' > /tmp/source.csv
OUTPUT_DIR=/tmp/csv-output SOURCE=/tmp/source.csv python copy_csv.py

The callback receives source as a Path and prefix as a string. Omitting PREFIX uses "copied"; setting PREFIX=flight writes flight.csv. Both names default to uppercase environment variables, so no lookups are needed inside the callback.

ctx.output_dir chooses where to write, and ctx.add_tabular tells Nominal how to read the result. The runtime writes manifest.json alongside your output. The timestamp options on add_tabular describe this output; the outer decorator’s timestamp options supply image-registration defaults. Importing copy_csv lets registration code call copy_csv.registration_kwargs() without running extraction.

run() is the container entrypoint. It reads the environment, calls your function, finalizes the output declaration, and on failure prints a traceback and exits non-zero so the ingest job fails.

Choose the import style that fits your code#

The examples use from nominal.experimental import extractor to make the framework namespace explicit. The same objects also live in focused public modules:

Module	Public API
`nominal.experimental.extractor.decorators`	`input`, `parameter`, `error`, `manifest_extractor`, `single_file_extractor`
`nominal.experimental.extractor.context`	`ExtractorContext`, `ManifestExtractorContext`, `SingleFileExtractorContext`
`nominal.experimental.extractor.types`	`Choice`, `IntRange`, `FloatRange`, `BadParameter`
`nominal.experimental.extractor.runner`	`Extractor`, the decorated entrypoint with execution and metadata-export methods

For explicit imports, use:

from nominal.experimental.extractor.context import ManifestExtractorContext
from nominal.experimental.extractor.decorators import input, manifest_extractor, parameter
from nominal.experimental.extractor.types import IntRange

Package-level imports continue to work. ExtractorError and TimestampMetadata are also available from the package; their definitions belong to nominal.core.

Inputs and parameters#

Now suppose the parser needs optional calibration, a partition count, and an on/off setting. Add a declaration for each value instead of growing a setup block inside the callback. The example below shows the signature; replace its body with your parsing and output code.

Keep the extractor decorator outermost. Each @input("recording") or @parameter("parts") names a function argument and defaults to the uppercase environment variable (RECORDING, PARTS). Override it with envvar="SOURCE"; name= and description= supply registration metadata. Input file_suffixes= supplies the registration file filters.

from pathlib import Path
from nominal.experimental import extractor

@extractor.manifest_extractor
@extractor.input("recording", file_suffixes=["flight"])
@extractor.input("calibration", default=None)
@extractor.parameter("parts", type=int, default=2)
@extractor.parameter("enabled", type=bool, default=True)
def split(
    ctx: extractor.ManifestExtractorContext,
    *,
    recording: Path,
    calibration: Path | None,
    parts: int,
    enabled: bool,
) -> None:
    ...  # Process the resolved arguments and declare outputs using ctx.

Arguments are required unless a decorator supplies a default. Optional inputs use default=None and arrive as Path | None; supplied input paths must exist as files. Parameter defaults are already-converted Python values and may be None. Keep defaults in decorators rather than the function signature. Supplied strings use type=. When omitted, a str, int, float, or bool default selects its converter; otherwise conversion defaults to str. An explicit converter always wins. An empty string is a supplied value, not a request for the default. Booleans accept true/false, yes/no, on/off, and 1/0, case-insensitively. Custom converters may raise ValueError or TypeError to reject a value. All declared arguments resolve before user code runs. Binding failures raise ExtractorError and participate in the same error mappings as extraction failures.

When registered metadata is present in the container’s environment, it is authoritative. Local runs without it resolve directly from the supplied environment. Parameter types and defaults are enforced by this runtime; the registration API currently stores neither as structured fields.

See Inputs and parameters for which of the two a value belongs in.

Declaration reference#

Option	Applies to	Meaning
First positional argument	Both decorators	Exact Python function argument to populate
`envvar=`	Both	Container environment variable; defaults to the uppercase argument name
`name=`	Both	Registration display name; defaults to the argument name
`description=`	Both	Optional registration description
`file_suffixes=`	Inputs	Registration suffix filters such as `["mcap", "csv"]`; empty accepts any suffix
`default=None`	Inputs	Allow an omitted input and inject `None`; other input defaults are unsupported
`type=`	Parameters	Convert a supplied string; infer from a basic default when omitted, otherwise use `str`
`default=`	Parameters	Inject this Python value when absent; omitting it makes the parameter required

The first callback argument receives the context. Declare the remaining arguments explicitly; keyword-only arguments are recommended. Decorated arguments cannot be positional-only, have function-signature defaults, or be supplied only through **kwargs. Every required callback argument after the context needs a declaration. Type annotations help readers and type checkers; they do not select a converter. Use type=int for a required integer, or default=2 to infer it.

Keep the outer extractor decorator above all argument decorators. Registration lists preserve the top-to-bottom order of declarations. Duplicate argument names or environment variables fail during decoration; display names must be unique within each category (inputs or parameters). Environment variable names must be valid identifiers; OUTPUT_DIR, NOMINAL_EXTRACTOR_INPUT_DIR, and names beginning with _NOMINAL_ are reserved.

At runtime, declared arguments match registered metadata by environment variable, not display name. An omitted optional input becomes None; a supplied path that is not a file fails. With parameter metadata present, a parameter absent from that metadata is a contract error even if its decorator specifies a default. Without metadata, local runs use the declared environment variables.

Requiredness is derived from the decorator default and included in registration:

a missing required input raises in add_containerized, before anything uploads;
a missing required parameter is checked by the extractor runtime during binding, before your function executes. Use @parameter("name") for a required value or provide a decorator default.

Conversion and validation#

Use the built-in converters to express common constraints directly:

@extractor.manifest_extractor
@extractor.input("source", file_suffixes=["csv"])
@extractor.parameter("mode", type=extractor.Choice(["fast", "precise"]), default="fast")
@extractor.parameter("parts", type=extractor.IntRange(min=1, max=64), default=2)
@extractor.parameter("gain", type=extractor.FloatRange(min=0.0), default=1.0)
@extractor.parameter("enabled", default=True)
def process(
    ctx: extractor.ManifestExtractorContext,
    *,
    source: Path,
    mode: str,
    parts: int,
    gain: float,
    enabled: bool,
) -> None:
    ...  # Apply these settings while parsing source and declaring outputs.

Converter	Accepted supplied values
`str`	Any string, including an empty string
`int`	Strings accepted by Python’s integer converter
`float`	Strings accepted by Python’s float converter, excluding NaN and infinity
`bool`	`true/false`, `yes/no`, `on/off`, `1/0`, case-insensitively
`Choice([...])`	An exact, case-sensitive match from a nonempty collection of unique strings
`IntRange(min=None, max=None)`	An integer within the inclusive bounds; either bound may be omitted
`FloatRange(min=None, max=None)`	A finite number within the inclusive bounds; rejects NaN and infinity
Custom callable	A string passed to your callable, returning the callback’s Python value

Defaults are already-converted values: use default=2, not default="2", with type=int. Basic built-in converters and the three constraints above validate non-None defaults when the parameter is declared. A float default must be finite; integer values are also accepted for float parameters. default=None makes a parameter optional without inferring a converter; use, for example, type=int, default=None and annotate its callback argument as int | None. Arbitrary custom converters are never called on defaults; the author must supply a valid Python value.

For application-specific validation, raise BadParameter with a message suitable for the user:

def positive_odd(value: str) -> int:
    result = int(value)
    if result <= 0 or result % 2 == 0:
        raise extractor.BadParameter("must be a positive odd integer")
    return result

# Use @extractor.parameter("parts", type=positive_odd, default=3) above the callback.

BadParameter becomes ExtractorError naming the argument and environment variable and including your message. Do not put raw supplied values or secrets in that message. Ordinary ValueError and TypeError also become ExtractorError, but their messages and supplied values are omitted from the diagnostic. Other converter exceptions propagate unchanged through normal extractor error handling; their messages can appear in tracebacks and mapped termination reports. Wrap third-party failures in ValueError or TypeError to sanitize them, or BadParameter with a message safe to display.

Job metadata#

Newer pipelines also inject job context, all optional and None or empty on a local run:

ctx.ingest_job_rid
ctx.dataset_rid
ctx.additional_tags            # tags the ingest request applies to all data from this run
ctx.job_timestamp_metadata     # what an output falls back to when it declares none of its own

ctx.dataset_rid and ctx.ingest_job_rid identify what the run is writing into, which is useful in log output.

Manifest extractors#

Once inputs are resolved, the parser can produce more than one kind of result. A flight recording, for example, might contain telemetry, events, and camera footage. Keep one callback and declare each result through ctx; the manifest tells Nominal how to ingest each file.

Register these images with the MANIFEST output format. Use one method per output format, each taking only the options that format uses. The runner writes manifest.json after your callback returns.

Method	For	Options
`add_tabular`	`.csv` / `.parquet` (and `.gz`)	`tag_columns`, `channel_prefix`, `units`, `timestamp_column`/`timestamp_type`
`add_avro_stream`	`.avro` / `.avro.gz` (Avro stream files)	`channel_prefix`, `units`, `timestamp_type`
`add_journal_json`	`.jsonl` / `.jsonl.gz`, ingested as logs	`timestamp_column`/`timestamp_type`
`add_video`	any supported video container	`channel` (required), `start` or `frame_timestamps`

The gaps are deliberate. Avro records carry their own channel, values, and tags, so no tag columns are needed, but their timestamps are bare numbers, so add_avro_stream still takes a timestamp_type saying how to read them. Log samples carry no tags and all land on one channel, so tag columns and a channel prefix would be silently dropped. Each method also checks the file extension its format requires, so a mismatch fails at the call rather than server-side after upload.

from pathlib import Path
from nominal.experimental import extractor

@extractor.manifest_extractor(
    default_timestamp_column="ts",
    default_timestamp_type="epoch_nanoseconds",
)
@extractor.input("recording", name="Recording", file_suffixes=["flight"])
@extractor.parameter("parts", name="Parts", type=int, default=2)
def split(ctx: extractor.ManifestExtractorContext, *, recording: Path, parts: int) -> None:
    decoded = read_my_format(recording)

    for i, chunk in enumerate(chunks_of(decoded.telemetry, parts)):
        part = ctx.output_dir / f"part_{i}.parquet"
        write_parquet(chunk, part)
        ctx.add_tabular(part, tag_columns={"vehicle": "veh_id"}, units={"pressure": "Pa"})

    events = ctx.output_dir / "events.jsonl"
    write_jsonl(decoded.events, events)          # each line needs a MESSAGE field
    ctx.add_journal_json(events, timestamp_column="ts", timestamp_type="epoch_microseconds")

    footage = ctx.output_dir / "front.mp4"
    decoded.camera.write_h264(footage)
    ctx.add_video(footage, channel="camera/front", start=decoded.started_at)

if __name__ == "__main__":
    split.run()

Save this entrypoint as extractor.py for the build and registration examples. read_my_format, chunks_of, write_parquet, and write_jsonl are application-provided parsing and writing helpers. A manifest extractor may emit telemetry, videos, or only videos.

Units#

units maps channel names to unit symbols on each tabular or Avro output. The runtime copies the mapping into that output’s manifest entry and preserves symbols as supplied. Omitting it, passing None, or passing an empty mapping writes an empty units map.

Declare units where you declare the output, since different files from the same extractor can have different channels and units. For example:

ctx.add_tabular(table, units={"pressure": "Pa", "temperature": "K"})
ctx.add_avro_stream(records, units={"voltage": "V"})

These maps travel in the output manifest; they are not input parameters or image-registration metadata.

Videos#

add_video needs a channel (the video becomes a channel on the dataset, alongside the telemetry) and exactly one of two ways to establish absolute time:

# the video's own presentation timestamps, offset from an absolute start
ctx.add_video(footage, channel="camera/front", start="2026-01-01T12:00:00Z")

# ...optionally scaled, when the media plays at a different rate than the camera recorded at
ctx.add_video(footage, channel="camera/front", start=started_at, true_frame_rate=59.94)

# or one absolute nanosecond timestamp per frame, when you have precise per-frame metadata
ctx.add_video(footage, channel="camera/front", frame_timestamps=[1_753_000_000_000_000_000, ...])

frame_timestamps is the one declaration that writes: the runtime serializes the sidecar file the platform expects (front.mp4 gets front.mp4.timestamps.json), so you never reproduce that format by hand.

Timestamps accept a datetime, an ISO 8601 string, or integer nanoseconds since the epoch.

Note

Video outputs require a recent version of the Nominal platform. An older ingest pipeline ignores them, and rejects a manifest whose only outputs are videos. The container is given an identical environment either way, so the runtime cannot detect which one is running it.

Per-output timestamps#

add_tabular and add_journal_json take timestamp_column and timestamp_type together; add_avro_stream takes the type alone, since the schema fixes which field holds the timestamps.

Omit them on a file to inherit the ingest’s timestamp metadata, and set them when one run emits files with different timestamp fields or units. Declaring them here is the most specific end of a precedence chain that runs from one output file up to the image’s registered defaults; Timestamps lays out the whole chain and which type to emit. ctx.job_timestamp_metadata exposes what was resolved, if your code needs to branch on it.

Single-file extractors#

The original contract, for images registered with PARQUET, CSV, or AVRO_STREAM: the pipeline ingests exactly one output file, parsed according to the registered format. Use it for images already registered that way; write new extractors as manifest extractors.

from pathlib import Path
from nominal.core.container_image import FileOutputFormat
from nominal.experimental import extractor

@extractor.single_file_extractor(
    output_format=FileOutputFormat.PARQUET,
    default_timestamp_column="ts",
    default_timestamp_type="epoch_nanoseconds",
)
@extractor.input("recording", file_suffixes=["flight"])
def convert(ctx: extractor.SingleFileExtractorContext, *, recording: Path) -> None:
    table = read_my_format(recording)
    out = ctx.output_dir / "converted.parquet"
    write_parquet(table, out)
    ctx.set_output(out)                          # declare it

if __name__ == "__main__":
    convert.run()                                # the container entrypoint

A second set_output call raises, since the registered format describes one file. Everything below applies to both contracts.

The decorator and the registered format must agree. If they do not (a @single_file_extractor running under an image registered as MANIFEST, or the reverse), run() fails at startup with an explicit error rather than producing an ingest that silently drops output.

Errors#

Rejections come back as ExtractorError when they are about the extractor’s own contract (a reserved file name, an output outside $OUTPUT_DIR, no outputs declared, a timestamp type the manifest cannot express) and as the ordinary argument errors the rest of the client raises (ValueError subclasses) when the arguments themselves are malformed, including a file extension the declared format cannot read, or half of a timestamp_column / timestamp_type pair.

An undeclared file left in $OUTPUT_DIR is a warning, not a failure: the pipeline reads only what the manifest names, so it will not be ingested, but a scratch file does not fail the run.

run() turns any failure into a non-zero exit so the ingest job fails cleanly. Pass exit=False to re-raise instead, which is how you drive it in tests.

Structured failures#

Declare each expected failure with @error below the outer extractor decorator, alongside @input and @parameter. It works with both manifest and single-file extractors. The framework catches matching exceptions from startup, argument binding, extraction, or output finalization, writes structured JSON to the termination log and stderr, and exits with the mapped status:

from pathlib import Path
from nominal.experimental import extractor

class MalformedRecordingError(ValueError):
    pass

@extractor.manifest_extractor(
    default_timestamp_column="ts",
    default_timestamp_type="epoch_nanoseconds",
)
@extractor.error(
    MalformedRecordingError,
    code="MALFORMED_INPUT",
    exit_code=64,
    message="The recording could not be decoded.",
)
@extractor.error(
    extractor.ExtractorError,
    code="EXTRACTOR_CONTRACT",
    exit_code=65,
    message="The extractor contract could not be satisfied.",
)
@extractor.input("recording", file_suffixes=["flight"])
@extractor.parameter("parts", type=int, default=2)
def convert(ctx: extractor.ManifestExtractorContext, *, recording: Path, parts: int) -> None:
    ...  # Parse recording; raise MalformedRecordingError for invalid content; declare outputs.

if __name__ == "__main__":
    convert.run()

The payload contains code, message (from str(exception)), and retryable (defaults to False; set retryable=True on @error to override). Stack declarations for different Exception subclasses; duplicate declarations for the same class are rejected. The outer extractor decorator must stay on top, but the relative order of @error, @input, and @parameter does not affect error matching. Use specific exception classes for expected failures so programming errors are not mislabeled. Subclasses match; the nearest registered class in Python’s method resolution order wins. Mapping ExtractorError intentionally includes startup configuration, argument binding, and output-contract failures. Use a separate application exception such as MalformedRecordingError for bad input contents, so a deployment fault does not receive a data-error code.

The default destination is /dev/termination-log. For local tests or a custom container mount, pass an explicit trusted path to run(termination_log_path=...). Environment variables, including TERMINATION_LOG_PATH, cannot change the destination. Writing the file is best-effort: if it fails, the framework still emits JSON to stderr and exits with the mapped code.

The JSON payload is limited to 4,096 UTF-8 bytes. Long messages are shortened on a Unicode boundary; the code and retryable flag are preserved. An error code that cannot fit even with an empty message is rejected when declaring @error. Stderr contains the structured JSON followed by the full traceback, retaining the original exception message for debugging. Exit codes must be integers from 1 through 255. Unmapped failures keep the traceback and exit status 1; exit=False re-raises without writing a termination message. Direct calls to the decorated function likewise propagate exceptions; reporting belongs to run().

message= on @error supplies static fallback text. It is optional for runtime-only use, and does not replace str(exception) in runtime reports. registration_kwargs() requires it for each mapping, because the platform needs a useful message when no termination message is available. It rejects reserved codes and conflicting fallbacks for the same exit code; identical fallbacks are combined.

Logging#

The extractor’s stdout and stderr land in a log dataset in your workspace, capped at 1 MiB per job. The runtime’s own log lines go through the nominal.experimental.extractor logger; run() configures logging when it is the entrypoint, so your own logging calls show up there too.

To trace the framework, configure logging before calling run():

import logging

logging.basicConfig(level=logging.INFO)
logging.getLogger("nominal.experimental.extractor").setLevel(logging.DEBUG)
convert.run()

DEBUG events describe context lookup sources, argument binding, supplied-versus-default parameter selection, callback invocation, output finalization, registration export, and error mapping and termination-log writes. Binding logs include argument and environment variable names, but never parameter values or defaults. Existing output logs include file names; exception reporting still includes the exception message and traceback. Your logging handlers must also allow DEBUG records if you configure handler-level filters.

Test before building#

Call run with an explicit environment and exit=False to exercise binding, your extraction logic, and output finalization together. This needs no Docker or Nominal connection. Create the input fixture and output directory first; use a real recording in your extractor’s format:

from pathlib import Path
from extractor import split

in_dir = Path("test-data")
out_dir = Path("test-output")
out_dir.mkdir(parents=True, exist_ok=True)
ctx = split.run(
    env={"OUTPUT_DIR": str(out_dir), "RECORDING": str(in_dir / "flight-42.flight")},
    exit=False,
)
print(ctx.build_manifest())

env replaces the environment, it does not merge into it, so it must carry everything your code reads: OUTPUT_DIR, the inputs, and required parameters. Omitted optional parameters use their decorator defaults. Nothing falls back to the ambient value: omitting OUTPUT_DIR fails with ExtractorError, raised to the caller under exit=False as above, or printed as a traceback before a non-zero exit under run()’s default exit=True.

In tests, assert output contents and ctx.build_manifest() for manifest extractors; for single-file extractors, assert the declared file’s contents. Also cover missing required values, invalid conversions, optional inputs, and your parser’s expected failures. run() returns the context, not your callback’s return value. The callback should declare outputs and return None. A direct split(ctx) call binds arguments but does not finalize outputs or report mapped failures; prefer run(env=..., exit=False) for tests of the complete runtime.

Build the image#

Your image needs the nominal package, whatever reads your format (pyarrow, ffmpeg, and so on), and an entrypoint that calls .run(). Install from the project’s own dependency manifest rather than hand-listing packages in the Dockerfile, so the image gets the versions the parser was tested against:

FROM python:3.12-slim
COPY --from=ghcr.io/astral-sh/uv:0.9.7 /uv /usr/local/bin/uv
WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev
COPY extractor.py ./
ENTRYPOINT ["/app/.venv/bin/python", "/app/extractor.py"]

This installs into /app/.venv, outside the home directory, which the runtime environment requires. Build it for linux/amd64 and save the tarball as described in Build requirements, then register it.

Migrate context-based extractors#

Earlier versions of the package read inputs and parameters through the context instead of declarations. Editors and type checkers can flag these legacy lookups through their @deprecated annotations. The annotations do not emit additional runtime warnings.

Existing ctx.input(), ctx.inputs, ctx.param(), and ctx.get_param() calls still work and emit one logging warning per run. Migrate lookups into declarations and callback arguments:

Existing lookup	Declaration	Callback argument
`ctx.input("RECORDING")`	`@extractor.input("recording", envvar="RECORDING")`	`recording: Path`
`ctx.param("MODE")`	`@extractor.parameter("mode", envvar="MODE")`	`mode: str`
`int(ctx.get_param("PARTS", "2"))`	`@extractor.parameter("parts", envvar="PARTS", type=int, default=2)`	`parts: int`
`ctx.get_param("LABEL")`	`@extractor.parameter("label", envvar="LABEL", default=None)`	`label: str \| None`

For unnamed ctx.input() or ctx.inputs discovery, declare each registered input explicitly using its actual environment variable. Lookups by display name must likewise map to the registered environment variable. Preserve existing names, descriptions, suffixes, and requiredness when moving manually maintained metadata into decorators.

Keep ctx as the first argument and leave output declarations and job metadata access unchanged. Add timestamp defaults to the outer decorator (and output_format for single-file extractors), then replace manual registration fields with **extractor_function.registration_kwargs(). Keep .run() as the container entrypoint.

Migration can be incremental: mixed functions still run, but generated registration includes only decorated arguments; no declarations yields empty input and parameter lists. Until every required dependency is declared, retain complete manually supplied registration metadata. Bare single-file decorators still run without an explicit format, but cannot generate registration metadata until that format and timestamp defaults are supplied.