# Write an extractor in Python

The [container contract](https://dev.nominal.io/extractors/contract.md) 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](https://dev.nominal.io/extractors/register.md).

> **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](https://click.palletsprojects.com/en/stable/arguments/) for a positional input and an [option](https://click.palletsprojects.com/en/stable/options/) for a named setting:

```python
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`:

```python
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:

```bash
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:

```python
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.

```python
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](https://dev.nominal.io/extractors/contract.md#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:

```python
@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:

```python
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:

```python
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`](https://dev.nominal.io/python/reference/experimental/extractors/#nominal.experimental.extractor.context.ManifestExtractorContext.add_tabular) | `.csv` / `.parquet` (and `.gz`) | `tag_columns`, `channel_prefix`, `units`, `timestamp_column`/`timestamp_type` |
| [`add_avro_stream`](https://dev.nominal.io/python/reference/experimental/extractors/#nominal.experimental.extractor.context.ManifestExtractorContext.add_avro_stream) | `.avro` / `.avro.gz` ([Avro stream files](https://dev.nominal.io/python/guides/datasets/avro-streams.md)) | `channel_prefix`, `units`, `timestamp_type` |
| [`add_journal_json`](https://dev.nominal.io/python/reference/experimental/extractors/#nominal.experimental.extractor.context.ManifestExtractorContext.add_journal_json) | `.jsonl` / `.jsonl.gz`, ingested as logs | `timestamp_column`/`timestamp_type` |
| [`add_video`](https://dev.nominal.io/python/reference/experimental/extractors/#nominal.experimental.extractor.context.ManifestExtractorContext.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.

```python
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:

```python
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:

```python
# 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](https://dev.nominal.io/extractors/timestamps-and-tags.md#extractor-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.

```python
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:

```python
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()`:

```python
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:

```python
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:

```dockerfile
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](https://dev.nominal.io/extractors/contract.md#runtime-environment) requires. Build it for `linux/amd64` and save the tarball as described in [Build requirements](https://dev.nominal.io/extractors/contract.md#build-requirements), then [register it](https://dev.nominal.io/extractors/register.md).

## 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.