Skip to content

Schema Module

This module contains schema definitions and validation functionality.

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,
)

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

Examples:

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

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/schema.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
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

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

name class-attribute instance-attribute 

name: str | None = None

Name for the run.

options class-attribute instance-attribute 

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

Options for running the app.