Application Module

This section documents the application components of the Nextmv Python SDK - Local experience.

application

Application module for interacting with local Nextmv applications.

This module provides functionality to interact with applications in Nextmv, including application management, running applications, and managing inputs.

CLASS	DESCRIPTION
DownloadURL	Result of getting a download URL.
PollingOptions	Options for polling when waiting for run results.
UploadURL	Result of getting an upload URL.
Application	Class for interacting with applications in Nextmv Cloud.

FUNCTION	DESCRIPTION
poll	Function to poll for results with configurable options.

Application dataclass

Application(src: str, description: Optional[str] = None)

A decision model that can be executed.

You can import the Application class directly from local:

from nextmv.local import Application

This class represents an application in Nextmv, providing methods to interact with the application, run it with different inputs, manage versions, instances, experiments, and more.

PARAMETER	DESCRIPTION
src	Source of the application, when initialized locally. An application's source typically refers to the directory containing the app.yaml manifest. TYPE: str
description	Description of the application. TYPE: Optional[str] DEFAULT: None

Examples:

>>> from nextmv.local import Application
>>> app = Application(src="path/to/app")
>>> # Retrieve an app's run result
>>> result = app.run_result("run-id")

description class-attribute instance-attribute

description: Optional[str] = None

Description of the application.

initialize classmethod

initialize(
    src: Optional[str] = None,
    description: Optional[str] = None,
    destination: Optional[str] = None,
) -> Application

Initialize a sample Nextmv application, locally.

This method will create a new application in the local file system. The application is a dir with the name given by src (it becomes the source of the app), under the location given by destination. If the destination parameter is not specified, the current working directory is used as default. This method will scaffold the application with the necessary files and directories to have an opinionated structure for your decision model. Once the application is initialized, you are encouraged to complete it with the decision model itself, so that the application can be run locally.

If the src parameter is not provided, a random name will be generated for the application.

PARAMETER	DESCRIPTION
src	Source (ID, name) of the application. Will be generated if not provided. TYPE: str DEFAULT: None
description	Description of the application. TYPE: str DEFAULT: None
destination	Destination directory where the application will be initialized. If not provided, the current working directory will be used. TYPE: str DEFAULT: None

RETURNS	DESCRIPTION
Application	The initialized application instance.

Source code in nextmv/nextmv/local/application.py

@classmethod
def initialize(
    cls,
    src: Optional[str] = None,
    description: Optional[str] = None,
    destination: Optional[str] = None,
) -> "Application":
    """
    Initialize a sample Nextmv application, locally.

    This method will create a new application in the local file system. The
    application is a dir with the name given by `src` (it becomes the
    _source_ of the app), under the location given by `destination`. If the
    `destination` parameter is not specified, the current working directory
    is used as default. This method will scaffold the application with the
    necessary files and directories to have an opinionated structure for
    your decision model. Once the application is initialized, you are
    encouraged to complete it with the decision model itself, so that the
    application can be run locally.

    If the `src` parameter is not provided, a random name will be generated
    for the application.

    Parameters
    ----------
    src : str, optional
        Source (ID, name) of the application. Will be generated if not
        provided.
    description : str, optional
        Description of the application.
    destination : str, optional
        Destination directory where the application will be initialized. If
        not provided, the current working directory will be used.

    Returns
    -------
    Application
        The initialized application instance.
    """

    destination_dir = os.getcwd() if destination is None else destination
    app_id = src if src is not None else safe_id("app")

    # Create the new directory with the given name.
    app_src = os.path.join(destination_dir, app_id)
    if os.path.exists(app_src):
        raise FileExistsError(f"destination dir for src already exists: {app_src}")

    os.makedirs(app_src, exist_ok=False)

    # Get the path to the initial app structure template.
    current_file_dir = os.path.dirname(os.path.abspath(__file__))
    initial_app_structure_path = os.path.join(current_file_dir, "..", "default_app")
    initial_app_structure_path = os.path.normpath(initial_app_structure_path)

    # Copy everything from initial_app_structure to the new directory.
    if os.path.exists(initial_app_structure_path):
        shutil.copytree(initial_app_structure_path, app_src, dirs_exist_ok=True)

    return cls(
        src=app_src,
        description=description,
    )

new_run

new_run(
    input: Union[
        Input, dict[str, Any], BaseModel, str
    ] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    options: Optional[
        Union[Options, dict[str, str]]
    ] = None,
    configuration: Optional[
        Union[RunConfiguration, dict[str, Any]]
    ] = None,
    json_configurations: Optional[dict[str, Any]] = None,
    input_dir_path: Optional[str] = None,
) -> str

Run the application locally with the provided input.

This method is the local equivalent to cloud.Application.new_run, which submits the input to Nextmv Cloud. This method runs the application locally using the src of the app.

Make sure that the src attribute is set on the Application class before running locally, as it is required by the method.

PARAMETER	DESCRIPTION
input	Input to use for the run. This can be a nextmv.Input object, dict, BaseModel or str. If nextmv.Input is used, and the input_format is either nextmv.InputFormat.JSON or nextmv.InputFormat.TEXT, then the input data is extracted from the .data property. If you want to work with nextmv.InputFormat.CSV_ARCHIVE or nextmv.InputFormat.MULTI_FILE, you should use the input_dir_path argument instead. This argument takes precedence over the input. If input_dir_path is specified, this function looks for files in that directory and tars them, to later be uploaded using the upload_large_input method. If both the input_dir_path and input arguments are provided, the input is ignored. When input_dir_path is specified, the configuration argument must also be provided. More specifically, the RunConfiguration.format.format_input.input_type parameter dictates what kind of input is being submitted to the Nextmv Cloud. Make sure that this parameter is specified when working with the following input formats: nextmv.InputFormat.CSV_ARCHIVE nextmv.InputFormat.MULTI_FILE When working with JSON or text data, use the input argument directly. In general, if an input is too large, it will be uploaded with the upload_large_input method. TYPE: Union[Input, dict[str, Any], BaseModel, str] DEFAULT: None
name	Name of the local run. TYPE: Optional[str] DEFAULT: None
description	Description of the local run. TYPE: Optional[str] DEFAULT: None
options	Options to use for the run. This can be a nextmv.Options object or a dict. If a dict is used, the keys must be strings and the values must be strings as well. If a nextmv.Options object is used, the options are extracted from the .to_cloud_dict() method. Note that specifying options overrides the input.options (if the input is of type nextmv.Input). TYPE: Optional[Union[Options, dict[str, str]]] DEFAULT: None
configuration	Configuration to use for the run. This can be a cloud.RunConfiguration object or a dict. If the object is used, then the .to_dict() method is applied to extract the configuration. TYPE: Optional[Union[RunConfiguration, dict[str, Any]]] DEFAULT: None
json_configurations	Optional configurations for JSON serialization. This is used to customize the serialization before data is sent. TYPE: Optional[dict[str, Any]] DEFAULT: None
input_dir_path	Path to a directory containing input files. This is useful for input formats like nextmv.InputFormat.CSV_ARCHIVE or nextmv.InputFormat.MULTI_FILE. If both input and input_dir_path are specified, the input is ignored, and the files in the directory are used instead. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
str	ID (run_id) of the local run that was executed.

RAISES	DESCRIPTION
ValueError	If the src property for the Application is not specified. If neither input nor input_dir_path is specified. If input_dir_path is specified but configuration is not provided.
FileNotFoundError	If the manifest.yaml file cannot be found in the specified src directory.

Examples:

>>> from nextmv.local import Application
>>> app = Application(id="my-app", src="/path/to/app")
>>> run_id = app.new_run(
...     input={"vehicles": [{"id": "v1"}]},
...     options={"duration": "10s"}
... )
>>> print(f"Local run completed with ID: {run_id}")

Source code in nextmv/nextmv/local/application.py

def new_run(
    self,
    input: Union[Input, dict[str, Any], BaseModel, str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    options: Optional[Union[Options, dict[str, str]]] = None,
    configuration: Optional[Union[RunConfiguration, dict[str, Any]]] = None,
    json_configurations: Optional[dict[str, Any]] = None,
    input_dir_path: Optional[str] = None,
) -> str:
    """
    Run the application locally with the provided input.

    This method is the local equivalent to `cloud.Application.new_run`,
    which submits the input to Nextmv Cloud. This method runs the
    application locally using the `src` of the app.

    Make sure that the `src` attribute is set on the `Application` class
    before running locally, as it is required by the method.

    Parameters
    ----------
    input: Union[Input, dict[str, Any], BaseModel, str]
        Input to use for the run. This can be a `nextmv.Input` object,
        `dict`, `BaseModel` or `str`.

        If `nextmv.Input` is used, and the `input_format` is either
        `nextmv.InputFormat.JSON` or `nextmv.InputFormat.TEXT`, then the
        input data is extracted from the `.data` property.

        If you want to work with `nextmv.InputFormat.CSV_ARCHIVE` or
        `nextmv.InputFormat.MULTI_FILE`, you should use the
        `input_dir_path` argument instead. This argument takes precedence
        over the `input`. If `input_dir_path` is specified, this function
        looks for files in that directory and tars them, to later be
        uploaded using the `upload_large_input` method. If both the
        `input_dir_path` and `input` arguments are provided, the `input`
        is ignored.

        When `input_dir_path` is specified, the `configuration` argument
        must also be provided. More specifically, the
        `RunConfiguration.format.format_input.input_type` parameter
        dictates what kind of input is being submitted to the Nextmv Cloud.
        Make sure that this parameter is specified when working with the
        following input formats:

        - `nextmv.InputFormat.CSV_ARCHIVE`
        - `nextmv.InputFormat.MULTI_FILE`

        When working with JSON or text data, use the `input` argument
        directly.

        In general, if an input is too large, it will be uploaded with the
        `upload_large_input` method.
    name: Optional[str]
        Name of the local run.
    description: Optional[str]
        Description of the local run.
    options: Optional[Union[Options, dict[str, str]]]
        Options to use for the run. This can be a `nextmv.Options` object
        or a dict. If a dict is used, the keys must be strings and the
        values must be strings as well. If a `nextmv.Options` object is
        used, the options are extracted from the `.to_cloud_dict()` method.
        Note that specifying `options` overrides the `input.options` (if
        the `input` is of type `nextmv.Input`).
    configuration: Optional[Union[RunConfiguration, dict[str, Any]]]
        Configuration to use for the run. This can be a
        `cloud.RunConfiguration` object or a dict. If the object is used,
        then the `.to_dict()` method is applied to extract the
        configuration.
    json_configurations: Optional[dict[str, Any]]
        Optional configurations for JSON serialization. This is used to
        customize the serialization before data is sent.
    input_dir_path: Optional[str]
        Path to a directory containing input files. This is useful for
        input formats like `nextmv.InputFormat.CSV_ARCHIVE` or
        `nextmv.InputFormat.MULTI_FILE`. If both `input` and
        `input_dir_path` are specified, the `input` is ignored, and the
        files in the directory are used instead.

    Returns
    -------
    str
        ID (`run_id`) of the local run that was executed.

    Raises
    ------
    ValueError
        If the `src` property for the `Application` is not specified.
        If neither `input` nor `input_dir_path` is specified.
        If `input_dir_path` is specified but `configuration` is not provided.
    FileNotFoundError
        If the manifest.yaml file cannot be found in the specified `src` directory.

    Examples
    --------
    >>> from nextmv.local import Application
    >>> app = Application(id="my-app", src="/path/to/app")
    >>> run_id = app.new_run(
    ...     input={"vehicles": [{"id": "v1"}]},
    ...     options={"duration": "10s"}
    ... )
    >>> print(f"Local run completed with ID: {run_id}")
    """

    self.__validate_input_dir_path_and_configuration(input_dir_path, configuration)

    if self.src is None:
        raise ValueError("`src` property for the `Application` must be specified to run the application locally")

    if input is None and input_dir_path is None:
        raise ValueError("Either `input` or `input_directory` must be specified")

    try:
        manifest = Manifest.from_yaml(self.src)
    except FileNotFoundError as e:
        raise FileNotFoundError(
            f"Could not find manifest.yaml in {self.src}. Maybe specify a different `src` dir?"
        ) from e

    input_data = None if input_dir_path else self.__extract_input_data(input)
    options_dict = self.__extract_options_dict(options, json_configurations)
    run_config_dict = self.__extract_run_config(input, configuration, input_dir_path)
    run_id = run(
        app_id=self.src,
        src=self.src,
        manifest=manifest,
        run_config=run_config_dict,
        name=name,
        description=description,
        input_data=input_data,
        inputs_dir_path=input_dir_path,
        options=options_dict,
    )

    return run_id

new_run_with_result

new_run_with_result(
    input: Union[
        Input, dict[str, Any], BaseModel, str
    ] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    run_options: Optional[
        Union[Options, dict[str, str]]
    ] = None,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
    configuration: Optional[
        Union[RunConfiguration, dict[str, Any]]
    ] = None,
    json_configurations: Optional[dict[str, Any]] = None,
    input_dir_path: Optional[str] = None,
    output_dir_path: Optional[str] = ".",
) -> RunResult

Submit an input to start a new local run of the application and poll for the result. This is a convenience method that combines the new_run and run_result_with_polling methods, applying polling logic to check when the local run succeeded.

This method is the local equivalent to cloud.Application.new_run_with_result, which submits the input to Nextmv Cloud. This method runs the application locally using the src of the app.

Make sure that the src attribute is set on the Application class before running locally, as it is required by the method.

PARAMETER	DESCRIPTION
input	Input to use for the run. This can be a nextmv.Input object, dict, BaseModel or str. If nextmv.Input is used, and the input_format is either nextmv.InputFormat.JSON or nextmv.InputFormat.TEXT, then the input data is extracted from the .data property. If you want to work with nextmv.InputFormat.CSV_ARCHIVE or nextmv.InputFormat.MULTI_FILE, you should use the input_dir_path argument instead. This argument takes precedence over the input. If input_dir_path is specified, this function looks for files in that directory and tars them, to later be uploaded using the upload_large_input method. If both the input_dir_path and input arguments are provided, the input is ignored. When input_dir_path is specified, the configuration argument must also be provided. More specifically, the RunConfiguration.format.format_input.input_type parameter dictates what kind of input is being submitted to the Nextmv Cloud. Make sure that this parameter is specified when working with the following input formats: nextmv.InputFormat.CSV_ARCHIVE nextmv.InputFormat.MULTI_FILE When working with JSON or text data, use the input argument directly. In general, if an input is too large, it will be uploaded with the upload_large_input method. TYPE: Union[Input, dict[str, Any], BaseModel, str] DEFAULT: None
name	Name of the local run. TYPE: Optional[str] DEFAULT: None
description	Description of the local run. TYPE: Optional[str] DEFAULT: None
run_options	Options to use for the run. This can be a nextmv.Options object or a dict. If a dict is used, the keys must be strings and the values must be strings as well. If a nextmv.Options object is used, the options are extracted from the .to_cloud_dict() method. Note that specifying options overrides the input.options (if the input is of type nextmv.Input). TYPE: Optional[Union[Options, dict[str, str]]] DEFAULT: None
polling_options	Options to use when polling for the run result. TYPE: PollingOptions DEFAULT: DEFAULT_POLLING_OPTIONS
configuration	Configuration to use for the run. This can be a cloud.RunConfiguration object or a dict. If the object is used, then the .to_dict() method is applied to extract the configuration. TYPE: Optional[Union[RunConfiguration, dict[str, Any]]] DEFAULT: None
json_configurations	Optional configurations for JSON serialization. This is used to customize the serialization before data is sent. TYPE: Optional[dict[str, Any]] DEFAULT: None
input_dir_path	Path to a directory containing input files. This is useful for input formats like nextmv.InputFormat.CSV_ARCHIVE or nextmv.InputFormat.MULTI_FILE. If both input and input_dir_path are specified, the input is ignored, and the files in the directory are used instead. TYPE: Optional[str] DEFAULT: None
output_dir_path	Path to a directory where non-JSON output files will be saved. This is required if the output is non-JSON. If the directory does not exist, it will be created. Uses the current directory by default. TYPE: Optional[str] DEFAULT: "."

RETURNS	DESCRIPTION
RunResult	Result of the run, including output.

RAISES	DESCRIPTION
ValueError	If the src property for the Application is not specified. If neither input nor inputs_dir_path is specified. If inputs_dir_path is specified but configuration is not provided.
FileNotFoundError	If the manifest.yaml file cannot be found in the specified src directory.

Examples:

>>> from nextmv.local import Application
>>> app = Application(id="my-app", src="/path/to/app")
>>> run_result = app.new_run_with_result(
...     input={"vehicles": [{"id": "v1"}]},
...     options={"duration": "10s"}
... )
>>> print(f"Local run completed with ID: {run_result.id}")

Source code in nextmv/nextmv/local/application.py

def new_run_with_result(
    self,
    input: Union[Input, dict[str, Any], BaseModel, str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    run_options: Optional[Union[Options, dict[str, str]]] = None,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
    configuration: Optional[Union[RunConfiguration, dict[str, Any]]] = None,
    json_configurations: Optional[dict[str, Any]] = None,
    input_dir_path: Optional[str] = None,
    output_dir_path: Optional[str] = ".",
) -> RunResult:
    """
    Submit an input to start a new local run of the application and poll
    for the result. This is a convenience method that combines the
    `new_run` and `run_result_with_polling` methods, applying
    polling logic to check when the local run succeeded.

    This method is the local equivalent to
    `cloud.Application.new_run_with_result`, which submits the input to
    Nextmv Cloud. This method runs the application locally using the `src`
    of the app.

    Make sure that the `src` attribute is set on the `Application` class
    before running locally, as it is required by the method.

    Parameters
    ----------
    input: Union[Input, dict[str, Any], BaseModel, str]
        Input to use for the run. This can be a `nextmv.Input` object,
        `dict`, `BaseModel` or `str`.

        If `nextmv.Input` is used, and the `input_format` is either
        `nextmv.InputFormat.JSON` or `nextmv.InputFormat.TEXT`, then the
        input data is extracted from the `.data` property.

        If you want to work with `nextmv.InputFormat.CSV_ARCHIVE` or
        `nextmv.InputFormat.MULTI_FILE`, you should use the
        `input_dir_path` argument instead. This argument takes precedence
        over the `input`. If `input_dir_path` is specified, this function
        looks for files in that directory and tars them, to later be
        uploaded using the `upload_large_input` method. If both the
        `input_dir_path` and `input` arguments are provided, the `input` is
        ignored.

        When `input_dir_path` is specified, the `configuration` argument
        must also be provided. More specifically, the
        `RunConfiguration.format.format_input.input_type` parameter
        dictates what kind of input is being submitted to the Nextmv Cloud.
        Make sure that this parameter is specified when working with the
        following input formats:

        - `nextmv.InputFormat.CSV_ARCHIVE`
        - `nextmv.InputFormat.MULTI_FILE`

        When working with JSON or text data, use the `input` argument
        directly.

        In general, if an input is too large, it will be uploaded with the
        `upload_large_input` method.
    name: Optional[str]
        Name of the local run.
    description: Optional[str]
        Description of the local run.
    run_options: Optional[Union[Options, dict[str, str]]]
        Options to use for the run. This can be a `nextmv.Options` object
        or a dict. If a dict is used, the keys must be strings and the
        values must be strings as well. If a `nextmv.Options` object is
        used, the options are extracted from the `.to_cloud_dict()` method.
        Note that specifying `options` overrides the `input.options` (if
        the `input` is of type `nextmv.Input`).
    polling_options: PollingOptions, default=_DEFAULT_POLLING_OPTIONS
        Options to use when polling for the run result.
    configuration: Optional[Union[RunConfiguration, dict[str, Any]]]
        Configuration to use for the run. This can be a
        `cloud.RunConfiguration` object or a dict. If the object is used,
        then the `.to_dict()` method is applied to extract the
        configuration.
    json_configurations: Optional[dict[str, Any]]
        Optional configurations for JSON serialization. This is used to
        customize the serialization before data is sent.
    input_dir_path: Optional[str]
        Path to a directory containing input files. This is useful for
        input formats like `nextmv.InputFormat.CSV_ARCHIVE` or
        `nextmv.InputFormat.MULTI_FILE`. If both `input` and
        `input_dir_path` are specified, the `input` is ignored, and the
        files in the directory are used instead.
    output_dir_path : Optional[str], default="."
        Path to a directory where non-JSON output files will be saved. This
        is required if the output is non-JSON. If the directory does not
        exist, it will be created. Uses the current directory by default.

    Returns
    -------
    RunResult
        Result of the run, including output.

    Raises
    ------
    ValueError
        If the `src` property for the `Application` is not specified. If
        neither `input` nor `inputs_dir_path` is specified. If
        `inputs_dir_path` is specified but `configuration` is not provided.
    FileNotFoundError
        If the manifest.yaml file cannot be found in the specified `src`
        directory.

    Examples
    --------
    >>> from nextmv.local import Application
    >>> app = Application(id="my-app", src="/path/to/app")
    >>> run_result = app.new_run_with_result(
    ...     input={"vehicles": [{"id": "v1"}]},
    ...     options={"duration": "10s"}
    ... )
    >>> print(f"Local run completed with ID: {run_result.id}")
    """

    run_id = self.new_run(
        input=input,
        name=name,
        description=description,
        options=run_options,
        configuration=configuration,
        json_configurations=json_configurations,
        input_dir_path=input_dir_path,
    )

    return self.run_result_with_polling(
        run_id=run_id,
        polling_options=polling_options,
        output_dir_path=output_dir_path,
    )

run_metadata

run_metadata(run_id: str) -> RunInformation

Get the metadata of a local run.

This method is the local equivalent to cloud.Application.run_metadata, which retrieves the metadata of a remote run in Nextmv Cloud. This method is used to get the metadata of a run that was executed locally using the new_run or new_run_with_result method.

Retrieves information about a run without including the run output. This is useful when you only need the run's status and metadata.

PARAMETER	DESCRIPTION
run_id	ID of the run to retrieve metadata for. TYPE: str

RETURNS	DESCRIPTION
RunInformation	Metadata of the run (run information without output).

RAISES	DESCRIPTION
ValueError	If the .nextmv/runs directory does not exist at the application source, or if the specified run ID does not exist.

Examples:

>>> metadata = app.run_metadata("run-789")
>>> print(metadata.metadata.status_v2)
StatusV2.succeeded

Source code in nextmv/nextmv/local/application.py

def run_metadata(self, run_id: str) -> RunInformation:
    """
    Get the metadata of a local run.

    This method is the local equivalent to
    `cloud.Application.run_metadata`, which retrieves the metadata of a
    remote run in Nextmv Cloud. This method is used to get the metadata of
    a run that was executed locally using the `new_run` or
    `new_run_with_result` method.

    Retrieves information about a run without including the run output.
    This is useful when you only need the run's status and metadata.

    Parameters
    ----------
    run_id : str
        ID of the run to retrieve metadata for.

    Returns
    -------
    RunInformation
        Metadata of the run (run information without output).

    Raises
    ------
    ValueError
        If the `.nextmv/runs` directory does not exist at the application
        source, or if the specified run ID does not exist.

    Examples
    --------
    >>> metadata = app.run_metadata("run-789")
    >>> print(metadata.metadata.status_v2)
    StatusV2.succeeded
    """

    runs_dir = os.path.join(self.src, ".nextmv", "runs")
    if not os.path.exists(runs_dir):
        raise ValueError(f"`.nextmv/runs` dir does not exist at app source: {self.src}")

    run_dir = os.path.join(runs_dir, run_id)
    if not os.path.exists(run_dir):
        raise ValueError(f"`{run_id}` run dir does not exist at: {runs_dir}")

    info_file = os.path.join(run_dir, f"{run_id}.json")
    if not os.path.exists(info_file):
        raise ValueError(f"`{info_file}` file does not exist at: {run_dir}")

    with open(info_file) as f:
        info_dict = json.load(f)

    info = RunInformation.from_dict(info_dict)

    return info

run_result

run_result(
    run_id: str, output_dir_path: Optional[str] = "."
) -> RunResult

Get the local result of a run.

This method is the local equivalent to cloud.Application.run_result, which retrieves the result of a remote run in Nextmv Cloud. This method is used to get the result of a run that was executed locally using the new_run or new_run_with_result method.

Retrieves the complete result of a run, including the run output.

PARAMETER	DESCRIPTION
run_id	ID of the run to get results for. TYPE: str
output_dir_path	Path to a directory where non-JSON output files will be saved. This is required if the output is non-JSON. If the directory does not exist, it will be created. Uses the current directory by default. TYPE: Optional[str] DEFAULT: "."

RETURNS	DESCRIPTION
RunResult	Result of the run, including output.

RAISES	DESCRIPTION
ValueError	If the .nextmv/runs directory does not exist at the application source, or if the specified run ID does not exist.

Examples:

>>> result = app.run_result("run-123")
>>> print(result.metadata.status_v2)
'succeeded'

Source code in nextmv/nextmv/local/application.py

def run_result(self, run_id: str, output_dir_path: Optional[str] = ".") -> RunResult:
    """
    Get the local result of a run.

    This method is the local equivalent to `cloud.Application.run_result`,
    which retrieves the result of a remote run in Nextmv Cloud. This method
    is used to get the result of a run that was executed locally using the
    `new_run` or `new_run_with_result` method.

    Retrieves the complete result of a run, including the run output.

    Parameters
    ----------
    run_id : str
        ID of the run to get results for.
    output_dir_path : Optional[str], default="."
        Path to a directory where non-JSON output files will be saved. This
        is required if the output is non-JSON. If the directory does not
        exist, it will be created. Uses the current directory by default.

    Returns
    -------
    RunResult
        Result of the run, including output.

    Raises
    ------
    ValueError
        If the `.nextmv/runs` directory does not exist at the application
        source, or if the specified run ID does not exist.

    Examples
    --------
    >>> result = app.run_result("run-123")
    >>> print(result.metadata.status_v2)
    'succeeded'
    """

    run_information = self.run_metadata(run_id=run_id)

    return self.__run_result(
        run_id=run_id,
        run_information=run_information,
        output_dir_path=output_dir_path,
    )

run_result_with_polling

run_result_with_polling(
    run_id: str,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
    output_dir_path: Optional[str] = ".",
) -> RunResult

Get the result of a local run with polling.

This method is the local equivalent to cloud.Application.run_result_with_polling, which retrieves the result of a remote run in Nextmv Cloud. This method is used to get the result of a run that was executed locally using the new_run or new_run_with_result method.

Retrieves the result of a run including the run output. This method polls for the result until the run finishes executing or the polling strategy is exhausted.

PARAMETER	DESCRIPTION
run_id	ID of the run to retrieve the result for. TYPE: str
polling_options	Options to use when polling for the run result. TYPE: PollingOptions DEFAULT: _DEFAULT_POLLING_OPTIONS
output_dir_path	Path to a directory where non-JSON output files will be saved. This is required if the output is non-JSON. If the directory does not exist, it will be created. Uses the current directory by default. TYPE: Optional[str] DEFAULT: "."

RETURNS	DESCRIPTION
RunResult	Complete result of the run including output data.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.
TimeoutError	If the run does not complete after the polling strategy is exhausted based on time duration.
RuntimeError	If the run does not complete after the polling strategy is exhausted based on number of tries.

Examples:

>>> from nextmv.cloud import PollingOptions
>>> # Create custom polling options
>>> polling_opts = PollingOptions(max_tries=50, max_duration=600)
>>> # Get run result with polling
>>> result = app.run_result_with_polling("run-123", polling_opts)
>>> print(result.output)
{'solution': {...}}

Source code in nextmv/nextmv/local/application.py

def run_result_with_polling(
    self,
    run_id: str,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
    output_dir_path: Optional[str] = ".",
) -> RunResult:
    """
    Get the result of a local run with polling.

    This method is the local equivalent to
    `cloud.Application.run_result_with_polling`, which retrieves the result
    of a remote run in Nextmv Cloud. This method is used to get the result
    of a run that was executed locally using the `new_run` or
    `new_run_with_result` method.

    Retrieves the result of a run including the run output. This method
    polls for the result until the run finishes executing or the polling
    strategy is exhausted.

    Parameters
    ----------
    run_id : str
        ID of the run to retrieve the result for.
    polling_options : PollingOptions, default=_DEFAULT_POLLING_OPTIONS
        Options to use when polling for the run result.
    output_dir_path : Optional[str], default="."
        Path to a directory where non-JSON output files will be saved. This
        is required if the output is non-JSON. If the directory does not
        exist, it will be created. Uses the current directory by default.

    Returns
    -------
    RunResult
        Complete result of the run including output data.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    TimeoutError
        If the run does not complete after the polling strategy is
        exhausted based on time duration.
    RuntimeError
        If the run does not complete after the polling strategy is
        exhausted based on number of tries.

    Examples
    --------
    >>> from nextmv.cloud import PollingOptions
    >>> # Create custom polling options
    >>> polling_opts = PollingOptions(max_tries=50, max_duration=600)
    >>> # Get run result with polling
    >>> result = app.run_result_with_polling("run-123", polling_opts)
    >>> print(result.output)
    {'solution': {...}}
    """

    def polling_func() -> tuple[Any, bool]:
        run_information = self.run_metadata(run_id=run_id)
        if run_information.metadata.status_v2 in {
            StatusV2.succeeded,
            StatusV2.failed,
            StatusV2.canceled,
        }:
            return run_information, True

        return None, False

    run_information = poll(polling_options=polling_options, polling_func=polling_func)

    return self.__run_result(
        run_id=run_id,
        run_information=run_information,
        output_dir_path=output_dir_path,
    )

run_visuals

run_visuals(run_id: str) -> None

Open the local run visuals in a web browser.

This method opens the visual representation of a locally executed run in the default web browser. It assumes that the run was executed locally using the new_run or new_run_with_result method and that the necessary visualization files are present.

If the run was correctly configured to produce visual assets, then the run will contain a visuals directory with one or more HTML files. Each file is opened in a new tab in the default web browser.

PARAMETER	DESCRIPTION
run_id	ID of the local run to visualize. TYPE: str

RAISES	DESCRIPTION
ValueError	If the .nextmv/runs directory does not exist at the application source, or if the specified run ID does not exist.

Source code in nextmv/nextmv/local/application.py

def run_visuals(self, run_id: str) -> None:
    """
    Open the local run visuals in a web browser.

    This method opens the visual representation of a locally executed run
    in the default web browser. It assumes that the run was executed locally
    using the `new_run` or `new_run_with_result` method and that
    the necessary visualization files are present.

    If the run was correctly configured to produce visual assets, then the
    run will contain a `visuals` directory with one or more HTML files.
    Each file is opened in a new tab in the default web browser.

    Parameters
    ----------
    run_id : str
        ID of the local run to visualize.

    Raises
    ------
    ValueError
        If the `.nextmv/runs` directory does not exist at the application
        source, or if the specified run ID does not exist.
    """

    runs_dir = os.path.join(self.src, ".nextmv", "runs")
    if not os.path.exists(runs_dir):
        raise ValueError(f"`.nextmv/runs` dir does not exist at app source: {self.src}")

    run_dir = os.path.join(runs_dir, run_id)
    if not os.path.exists(run_dir):
        raise ValueError(f"`{run_id}` run dir does not exist at: {runs_dir}")

    visuals_dir = os.path.join(run_dir, "visuals")
    if not os.path.exists(visuals_dir):
        raise ValueError(f"`visuals` dir does not exist at: {run_dir}")

    for file in os.listdir(visuals_dir):
        if file.endswith(".html"):
            file_path = os.path.join(visuals_dir, file)
            webbrowser.open_new_tab(f"file://{os.path.realpath(file_path)}")

src instance-attribute

src: str

Source of the application, when initialized locally. An application's source typically refers to the directory containing the app.yaml manifest.

sync

sync(
    target: Application,
    run_ids: Optional[list[str]] = None,
    instance_id: Optional[str] = None,
    verbose: Optional[bool] = False,
) -> None

Sync the local application to a Nextmv Cloud application target.

The Application class allows you to perform and handle local application runs with methods such as:

• new_run
• new_run_with_result
• run_metadata
• run_result
• run_result_with_polling

The runs produced locally live under self.src/.nextmv/runs. This method syncs those runs to a Nextmv Cloud application target, making them available for remote execution and management.

PARAMETER	DESCRIPTION
target	Target Nextmv Cloud application where the local application runs will be synced to. TYPE: Application
run_ids	List of run IDs to sync. If None, all local runs found under self.src/.nextmv/runs will be synced. TYPE: Optional[list[str]] DEFAULT: None
instance_id	Optional instance ID if you want to associate your runs with an instance. TYPE: Optional[str] DEFAULT: None
verbose	Whether to print verbose output during the sync process. Useful for debugging a large number of runs being synced. TYPE: Optional[bool] DEFAULT: False

RAISES	DESCRIPTION
ValueError	If the src property is not specified.
ValueError	If the client property is not specified.
ValueError	If the application does not exist in Nextmv Cloud.
ValueError	If a run does not exist locally.
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/local/application.py

def sync(  # noqa: C901
    self,
    target: cloud.Application,
    run_ids: Optional[list[str]] = None,
    instance_id: Optional[str] = None,
    verbose: Optional[bool] = False,
) -> None:
    """
    Sync the local application to a Nextmv Cloud application target.

    The `Application` class allows you to perform and handle local
    application runs with methods such as:

    - `new_run`
    - `new_run_with_result`
    - `run_metadata`
    - `run_result`
    - `run_result_with_polling`

    The runs produced locally live under `self.src/.nextmv/runs`. This
    method syncs those runs to a Nextmv Cloud application target, making
    them available for remote execution and management.

    Parameters
    ----------
    target : cloud.Application
        Target Nextmv Cloud application where the local application runs
        will be synced to.
    run_ids : Optional[list[str]], default=None
        List of run IDs to sync. If None, all local runs found under
        `self.src/.nextmv/runs` will be synced.
    instance_id : Optional[str], default=None
        Optional instance ID if you want to associate your runs with an
        instance.
    verbose : Optional[bool], default=False
        Whether to print verbose output during the sync process. Useful for
        debugging a large number of runs being synced.

    Raises
    ------
    ValueError
        If the `src` property is not specified.
    ValueError
        If the `client` property is not specified.
    ValueError
        If the application does not exist in Nextmv Cloud.
    ValueError
        If a run does not exist locally.
    requests.HTTPError
        If the response status code is not 2xx.
    """
    if self.src is None:
        raise ValueError(
            "`src` property for the `Application` must be specified to sync the application to Nextmv Cloud"
        )

    if target.client is None:
        raise ValueError(
            "`client` property for the target `Application` must be specified to sync the application to Cloud"
        )

    if not target.exists(target.client, target.id):
        raise ValueError(
            "target Application does not exist in Nextmv Cloud, create it with `cloud.Application.new`"
        )
    if verbose:
        log(f"☁️ Starting sync of local application `{self.src}` to Nextmv Cloud application `{target.id}`.")

    # Create a temp dir to store the outputs that are written by default to
    # ".". During the sync process, we don't need to keep these outputs, so
    # we can use a temp dir that will be deleted after the sync is done.
    with tempfile.TemporaryDirectory(prefix="nextmv-sync-run-") as temp_results_dir:
        runs_dir = os.path.join(self.src, ".nextmv", "runs")
        if run_ids is None:
            # If runs are not specified, by default we sync all local runs that
            # can be found.
            dirs = os.listdir(runs_dir)
            run_ids = [d for d in dirs if os.path.isdir(os.path.join(runs_dir, d))]

            if verbose:
                log(f"ℹ️  Found {len(run_ids)} local runs to sync from {runs_dir}.")
        else:
            if verbose:
                log(f"ℹ️  Syncing {len(run_ids)} specified local runs from {runs_dir}.")

        total = 0
        for run_id in run_ids:
            synced = self.__sync_run(
                target=target,
                run_id=run_id,
                runs_dir=runs_dir,
                temp_dir=temp_results_dir,
                instance_id=instance_id,
                verbose=verbose,
            )
            if synced:
                total += 1

        if verbose:
            log(
                f"🚀 Process completed, synced local application `{self.src}` to "
                f"Nextmv Cloud application `{target.id}`: "
                f"{total}/{len(run_ids)} runs."
            )