Skip to content

Schema module

Reference to the schema.py module.

schema

Schema definitions for Nextpipe.

This module contains schema definitions used for pipeline configurations.

CLASS DESCRIPTION
AppOption

Option for running an app.
AppRunConfig

Configuration for running an app.

AppOption dataclass 

AppOption(name: str, value: Any)

Option for running an app.

You can import the AppOption class directly from nextpipe:

from nextpipe import AppOption

This class represents a key-value pair for specifying options when running an app in a pipeline.

PARAMETER DESCRIPTION

name

Key for the option.

TYPE: str

value

Value for the option.

TYPE: Any

Examples:

>>> from nextpipe import AppOption
>>> option = AppOption(name="threads", value=4)

name instance-attribute 

name: str

Key for the option.

value instance-attribute 

value: Any

Value for the option.

AppRunConfig dataclass 

AppRunConfig(
    input: dict[str, Any] | str | Any,
    options: list[AppOption] | dict[str, Any] = list(),
    name: str | None = None,
    description: str | None = None,
    run_configuration: RunConfiguration | None = None,
    instance_id: str | None = None,
)

Configuration for running an app.

You can import the AppRunConfig class directly from nextpipe:

from nextpipe import AppRunConfig

This class represents a configuration object used when running an app in a pipeline, containing input data, options, and an optional name.

PARAMETER DESCRIPTION

input

Input data for the app. A JSON app can take a dictionary, multi-file apps can take a directory path as a string. Other types will be passed to the underlying Python SDK as-is (e.g., nextmv.Input).

TYPE: Union[dict[str, Any], str, Any]

options

Options for running the app, by default empty. These can be provided as a list of AppOption instances, or, simply as a dictionary of key-value pairs.

TYPE: Union[list[AppOption], dict[str, Any]] DEFAULT: list()

name

Name for the run, by default None.

TYPE: str DEFAULT: None

description

Description for the run, by default None.

TYPE: str DEFAULT: None

run_configuration

The configuration to apply when running the app, by default None. If given, it replaces the run_configuration of the app decorator for this run. Note that it replaces it as a whole, i.e., no fields are inherited from the decorator's configuration.

TYPE: RunConfiguration DEFAULT: None

instance_id

The ID of the instance to run, by default None. If given, it replaces the instance_id of the app decorator for this run.

TYPE: str DEFAULT: None

Examples:

>>> from nextpipe import AppRunConfig, AppOption
>>> config = AppRunConfig(
...     input={"data": [1, 2, 3]},
...     options={"threads": 4},
...     name="my-run"
... )

The run configuration and instance can be chosen per run, e.g., to pick an execution class based on the size of the scenario at hand.

>>> import nextmv
>>> from nextpipe import AppRunConfig
>>> config = AppRunConfig(
...     input={"data": [1, 2, 3]},
...     run_configuration=nextmv.RunConfiguration(execution_class="8c16gb12h"),
...     instance_id="tuned-for-large",
... )

input instance-attribute 

input: dict[str, Any] | str | Any

Input for the app. A JSON app can take a dictionary, multi-file apps can take a directory path as a string. Other types will be passed to the underlying Python SDK as-is (e.g., nextmv.Input).

options class-attribute instance-attribute 

options: list[AppOption] | dict[str, Any] = field(
    default_factory=list
)

Options for running the app.

name class-attribute instance-attribute 

name: str | None = None

Name for the run.

description class-attribute instance-attribute 

description: str | None = None

Description for the run.

run_configuration class-attribute instance-attribute 

run_configuration: RunConfiguration | None = field(
    default=None,
    metadata=config(
        encoder=_encode_run_configuration,
        decoder=_decode_run_configuration,
    ),
)

The configuration to apply when running the app. Replaces the run_configuration of the app decorator as a whole, if given.

instance_id class-attribute instance-attribute 

instance_id: str | None = None

The ID of the instance to run. Replaces the instance_id of the app decorator, if given.

get_options 

get_options() -> dict[str, Any]

Get options as a dictionary.

This method converts the options attribute to a dictionary if it is provided as a list of AppOption instances.

RETURNS DESCRIPTION
dict[str, Any]

Dictionary of options.
Source code in nextpipe/nextpipe/schema.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def get_options(self) -> dict[str, Any]:
    """
    Get options as a dictionary.

    This method converts the `options` attribute to a dictionary if it is provided
    as a list of `AppOption` instances.

    Returns
    -------
    dict[str, Any]
        Dictionary of options.
    """
    options = self.options
    if isinstance(self.options, list):
        options = {option.name: option.value for option in self.options}
    options = utils.convert_to_string_values(options)
    return options