single_file_extractor

nominal.experimental.extractor.decorators.single_file_extractor(fn: Callable[[...], None], *, default_timestamp_column: str | None = None, default_timestamp_type: Iso8601 | Epoch | Relative | Custom | Literal['iso_8601', 'epoch_picoseconds', 'epoch_nanoseconds', 'epoch_microseconds', 'epoch_milliseconds', 'epoch_seconds', 'epoch_minutes', 'epoch_hours', 'epoch_days'] | None = None, output_format: FileOutputFormat | None = None) → Extractor[SingleFileExtractorContext]

nominal.experimental.extractor.decorators.single_file_extractor(*, default_timestamp_column: str | None = None, default_timestamp_type: Iso8601 | Epoch | Relative | Custom | Literal['iso_8601', 'epoch_picoseconds', 'epoch_nanoseconds', 'epoch_microseconds', 'epoch_milliseconds', 'epoch_seconds', 'epoch_minutes', 'epoch_hours', 'epoch_days'] | None = None, output_format: FileOutputFormat | None = None) → Callable[[Callable[[...], None]], Extractor[SingleFileExtractorContext]]

Declare a single-file extractor with ctx followed by injected arguments.

Use @error(ExceptionType, code="...", exit_code=64) below this decorator to declare structured failures handled automatically by Extractor.run(). Declare inputs and parameters below this decorator with @input and @parameter; they are passed as keyword arguments after ctx. Declare default_timestamp_column and default_timestamp_type to generate Extractor.registration_kwargs().

For images registered with a single-file output format (PARQUET, CSV, …): the ingest pipeline ingests exactly one output file, parsed per the registered format. Declare it with SingleFileExtractorContext.set_output(). If the image’s registered format turns out to be MANIFEST, Extractor.run() fails at startup with a clear error.

Set output_format explicitly to generate registration metadata; it must then match the registered format exactly. Omitting it preserves legacy runtime behavior.

This is the original output contract, kept for images already registered against it. New extractors should use manifest_extractor(), a strict superset: this mode has no per-output timestamp, tag-column, or channel-prefix control, and changing an image’s output format later requires registering a new image.

Example

from pathlib import Path

from nominal.core.container_image import FileOutputFormat
from nominal.experimental.extractor import SingleFileExtractorContext, input, single_file_extractor

@single_file_extractor(
    output_format=FileOutputFormat.PARQUET,
    default_timestamp_column="time_us",
    default_timestamp_type="epoch_microseconds",
)
@input("source")
def convert(ctx: SingleFileExtractorContext, *, source: Path) -> None:
    table = read_input(source)
    out = ctx.output_dir / "converted.parquet"
    write_parquet(table, out)
    ctx.set_output(out)

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