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
Application	Class for interacting with local Nextmv applications.

Application

Bases: BaseModel

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

Returned by:

• nextmv 💻 Local Reference Application Module  application  Application
  •  from_registry
  •  get_or_register
  •  initialize

app_id class-attribute instance-attribute

app_id: str | None = None

ID of the application. This ID is auto-generated if not provided.

content_format class-attribute instance-attribute

content_format: ContentFormat | InputFormat | None = None

Warning

InputFormat is deprecated, but kept for backward compatibility. Use ContentFormat instead.

The content format of the application, which is determined by the manifest.

description class-attribute instance-attribute

description: str | None = None

Description of the application.

from_registry classmethod

from_registry(
    src: str | None = None, app_id: str | None = None
) -> Application

Load an application from the local registry.

The local registry is a YAML file stored at $HOME/.nextmv/registry.yaml that keeps track of locally registered applications. This method looks for an entry in the local registry that matches the provided src or app_id, and loads the application from the corresponding source path.

Specify either the src or the app_id to identify the application to load. If both are provided, the method will prioritize loading by app_id.

PARAMETER	DESCRIPTION
src	Source path of the application to load. This is typically the path to the directory containing the application's manifest (app.yaml). TYPE: str | None DEFAULT: None
app_id	ID of the application to load. This ID is auto-generated if not provided during registration. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
Application	The loaded application instance.

RAISES	DESCRIPTION
ValueError	If neither src nor app_id is provided, or if no matching entry is found in the registry.

Examples:

Assuming an application is registered in the local registry with the following entry:

apps:
  - app_id: "app-123"
    src: "/path/to/app"

>>> from nextmv.local import Application
>>> app = Application.from_registry(app_id="app-123")
>>> print(app.src)
/path/to/app

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

@classmethod
def from_registry(cls, src: str | None = None, app_id: str | None = None) -> "Application":
    """
    Load an application from the local registry.

    The local registry is a YAML file stored at `$HOME/.nextmv/registry.yaml`
    that keeps track of locally registered applications. This method looks
    for an entry in the local registry that matches the provided `src` or
    `app_id`, and loads the application from the corresponding source path.

    Specify either the `src` or the `app_id` to identify the application to
    load. If both are provided, the method will prioritize loading by
    `app_id`.

    Parameters
    ----------
    src : str | None
        Source path of the application to load. This is typically the path
        to the directory containing the application's manifest (`app.yaml`).
    app_id : str | None
        ID of the application to load. This ID is auto-generated if not
        provided during registration.

    Returns
    -------
    Application
        The loaded application instance.

    Raises
    ------
    ValueError
        If neither `src` nor `app_id` is provided, or if no matching entry is found in the registry.

    Examples
    --------
    Assuming an application is registered in the local registry with the following entry:

    ```yaml
    apps:
      - app_id: "app-123"
        src: "/path/to/app"
    ```

    >>> from nextmv.local import Application
    >>> app = Application.from_registry(app_id="app-123")
    >>> print(app.src)
    /path/to/app
    """

    if src is None and app_id is None:
        raise ValueError("Either `src` or `app_id` must be provided to load an application from the registry")

    reg = Registry.from_yaml()
    entry = reg.entry(app_id=app_id, src=src)

    if entry is None:
        raise ValueError("No matching entry found in the registry for the provided `src` or `app_id`")

    # Validate that the app.yaml file exists in the specified source
    # directory before loading the application.
    try:
        Manifest.from_yaml(entry.src)
    except FileNotFoundError as e:
        raise FileNotFoundError(
            f"Could not find app.yaml in `{entry.src}`. Maybe specify a different `src` dir?"
        ) from e

    if not os.path.exists(entry.src):
        raise FileNotFoundError(f"App was found in the registry but the path does not exist: {entry.src}")

    return cls(
        src=entry.src,
        app_id=entry.app_id,
        description=entry.description,
    )

get_or_register classmethod

get_or_register(
    app_src: str, app_id: str | None = None
) -> tuple[Application, bool]

Get a local application from the registry or register it if it does not exist.

This function checks if a local application with the given source and ID is already registered in the local registry. If it is, it retrieves and returns that application. If it is not registered, it creates a new Application instance with the provided source and ID, registers it in the local registry, and then returns it. The boolean in the returned tuple indicates whether a new application was registered (True) or an existing one was retrieved (False).

PARAMETER	DESCRIPTION
app_src	The source path of the application to get or register. TYPE: str
app_id	The ID of the application to get or register. If None, the ID will be derived from the application source. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
tuple[Application, bool]	A tuple containing the local application retrieved from the registry or newly registered, and a boolean indicating whether the application was newly registered.

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

@classmethod
def get_or_register(cls, app_src: str, app_id: str | None = None) -> tuple["Application", bool]:
    """
    Get a local application from the registry or register it if it does not
    exist.

    This function checks if a local application with the given source and
    ID is already registered in the local registry. If it is, it retrieves
    and returns that application. If it is not registered, it creates a new
    `Application` instance with the provided source and ID, registers it in
    the local registry, and then returns it. The boolean in the returned
    tuple indicates whether a new application was registered (True) or an
    existing one was retrieved (False).

    Parameters
    ----------
    app_src : str
        The source path of the application to get or register.
    app_id : Optional[str]
        The ID of the application to get or register. If None, the ID will be
        derived from the application source.

    Returns
    -------
    tuple[Application, bool]
        A tuple containing the local application retrieved from the registry or
        newly registered, and a boolean indicating whether the application was
        newly registered.
    """

    app = cls(src=app_src, app_id=app_id)
    if app.is_registered():
        return cls.from_registry(src=app_src, app_id=app_id), False

    app.register()

    return app, True

initialize classmethod

initialize(
    src: str | None = None,
    description: str | None = None,
    destination: str | None = None,
    manifest_type: ManifestType = PYTHON,
    content_format: ContentFormat = JSON,
    example: str | None = "hello-world",
    should_register: bool = False,
) -> Application

Initialize a sample Nextmv application, locally.

This method will create a new application from a template 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. The template used is determined by the combination of manifest_type and content_format. The example parameter further specifies which example within the template to use when scaffolding the application. 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. If should_register is set to True, the application will also be registered in the local registry after initialization.

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
manifest_type	Type of manifest to use for the application. Determines the language/runtime template that will be scaffolded. TYPE: ManifestType DEFAULT: ManifestType.PYTHON
content_format	Content format of the application's input and output. Determines which template variant is used when scaffolding the application. TYPE: ContentFormat DEFAULT: ContentFormat.JSON
example	The example template to use when initializing the application. Valid values depend on the manifest_type. - PYTHON: hello-world, class-assign, demand-alloc - GO: hello-world - JAVA: hello-world - BINARY: hello-world TYPE: str DEFAULT: "hello-world"
should_register	Whether to register the application in the local registry after initialization. If True, the application will be added to $HOME/.nextmv/registry.yaml. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
Application	The initialized application instance.

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

@classmethod
def initialize(
    cls,
    src: str | None = None,
    description: str | None = None,
    destination: str | None = None,
    manifest_type: ManifestType = ManifestType.PYTHON,
    content_format: ContentFormat = ContentFormat.JSON,
    example: str | None = "hello-world",
    should_register: bool = False,
) -> "Application":
    """
    Initialize a sample Nextmv application, locally.

    This method will create a new application from a template 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. The template used is
    determined by the combination of `manifest_type` and `content_format`.
    The `example` parameter further specifies which example within the
    template to use when scaffolding the application. 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. If `should_register` is set to True, the
    application will also be registered in the local registry after
    initialization.

    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.
    manifest_type : ManifestType, default=ManifestType.PYTHON
        Type of manifest to use for the application. Determines the
        language/runtime template that will be scaffolded.
    content_format : ContentFormat, default=ContentFormat.JSON
        Content format of the application's input and output. Determines
        which template variant is used when scaffolding the application.
    example : str, optional, default="hello-world"
        The example template to use when initializing the application.
        Valid values depend on the `manifest_type`.
        - `PYTHON`: `hello-world`, `class-assign`, `demand-alloc`
        - `GO`: `hello-world`
        - `JAVA`: `hello-world`
        - `BINARY`: `hello-world`
    should_register : bool, default=False
        Whether to register the application in the local registry after
        initialization. If True, the application will be added to
        `$HOME/.nextmv/registry.yaml`.

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

    destination_dir = os.getcwd() if not destination else destination
    app_id = src if src else safe_id("local-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}")

    example = example or "hello-world"

    # Validate that the example and manifest_type combo are valid.
    if example not in {"hello-world", "class-assign", "demand-alloc"}:
        raise ValueError(
            f"Invalid example: {example}. Valid examples are 'hello-world', 'class-assign' and 'demand-alloc'."
        )

    if example != "hello-world" and manifest_type != ManifestType.PYTHON:
        raise ValueError(
            "Example templates other than 'hello-world' are only available for Python. "
            f"Received manifest type: {manifest_type.value}"
        )

    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__))
    template = f"{manifest_type.value}_{content_format.value}_{example}"
    # Note: we are deliberately including the templates directory in the pyinstaller
    # release, as it otherwise would not be included and the initialize function would
    # break. Make sure to keep this in mind if changing the location of the templates.
    initial_app_structure_path = os.path.join(current_file_dir, "..", "templates", template)
    initial_app_structure_path = os.path.normpath(initial_app_structure_path)

    # Copy everything from initial_app_structure to the new directory.
    if not os.path.exists(initial_app_structure_path):
        raise FileNotFoundError(f"Initial app structure template not found at: {initial_app_structure_path}")

    shutil.copytree(initial_app_structure_path, app_src, dirs_exist_ok=True)

    # Workaround to provide a sample manifest when initializing with the
    # demand allocation example, as this example is a folder of two
    # applications, instead of a standalone app.
    manifest = None
    if example == "demand-alloc":
        manifest = Manifest.from_yaml(os.path.join(app_src, "workflow"))

    local_app = cls(
        src=app_src,
        description=description,
        content_format=content_format,
        manifest=manifest,
    )

    # To clean up the workaround mentioned above, we now remove the
    # manifest file from the dir.
    if example == "demand-alloc":
        manifest_path = os.path.join(app_src, MANIFEST_FILE_NAME)
        if os.path.exists(manifest_path):
            os.remove(manifest_path)

    if should_register:
        local_app.register()

    return local_app

is_registered

is_registered() -> bool

Check if the local application is registered in the local registry.

This method checks if the application has an entry in the local registry.

RETURNS	DESCRIPTION
bool	True if the application is registered, False otherwise.

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

def is_registered(self) -> bool:
    """
    Check if the local application is registered in the local registry.

    This method checks if the application has an entry in the local registry.

    Returns
    -------
    bool
        True if the application is registered, False otherwise.
    """

    reg = Registry.from_yaml()
    entry = reg.entry(app_id=self.app_id, src=self.src)

    return entry is not None

list_runs

list_runs(status: StatusV2 | None = None) -> list[Run]

List all runs for the application.

PARAMETER	DESCRIPTION
status	If specified, only runs with the given status will be returned. TYPE: Optional[StatusV2] DEFAULT: None

RETURNS	DESCRIPTION
list[Run]	A list of all runs associated with the application.

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

def list_runs(self, status: StatusV2 | None = None) -> list[Run]:
    """
    List all runs for the application.

    Parameters
    ----------
    status : Optional[StatusV2]
        If specified, only runs with the given status will be returned.

    Returns
    -------
    list[Run]
        A list of all runs associated with the application.
    """

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

    dirs = os.listdir(runs_dir)
    if not dirs:
        return []

    run_ids = [d for d in dirs if os.path.isdir(os.path.join(runs_dir, d))]
    if not run_ids:
        return []

    runs = []
    for run_id in run_ids:
        try:
            info = self.run_metadata(run_id=run_id)
        except (ValueError, FileNotFoundError, json.JSONDecodeError, OSError) as e:
            # Skip runs with expected errors (missing/corrupted files)
            log(f"Warning: Skipping run '{run_id}' due to error: {e}")
            continue

        run = info.to_run()
        if status is not None and run.status_v2 != status:
            continue

        runs.append(run)

    return runs

manifest class-attribute instance-attribute

manifest: Manifest | None = Field(
    default=None, exclude=True
)

Manifest of the application. A manifest is a file named app.yaml that must be present at the root of the application's src directory. If you specify this argument when creating the Application, the provided manifest will be written to the src directory, overriding any existing manifest file.

model_post_init

model_post_init(__context) -> None

Actions taken after the Application model is initialized.

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

def model_post_init(self, __context) -> None:
    """
    Actions taken after the Application model is initialized.
    """

    # If this application is already registered, we use the information in
    # the registry to normalize some of the attributes of the class.
    reg = Registry.from_yaml()
    entry = reg.entry(app_id=self.app_id, src=self.src)
    if entry is not None:
        self.app_id = entry.app_id
        self.src = entry.src
        self.description = entry.description
        self.content_format = entry.content_format

    if self.manifest is not None:
        self.manifest.to_yaml(self.src)

    try:
        manifest = Manifest.from_yaml(self.src)
        self.manifest = manifest

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

    content_format = ContentFormat.JSON
    if self.manifest.configuration is not None and self.manifest.configuration.content is not None:
        content_format = self.manifest.configuration.content.format

    # Translate deprecated `InputFormat` to `ContentFormat` for backwards
    # compatibility.
    if type(content_format) is InputFormat:
        if content_format == InputFormat.JSON:
            content_format = ContentFormat.JSON
        elif content_format == InputFormat.MULTI_FILE:
            content_format = ContentFormat.MULTI_FILE

    self.content_format = content_format

    # Always normalize the source path to an absolute path, to avoid issues
    # with relative paths.
    self.src = os.path.abspath(self.src)

new_run

new_run(
    input: Input | dict[str, Any] | BaseModel | str = None,
    name: str | None = None,
    description: str | None = None,
    options: Options | dict[str, str] | None = None,
    configuration: RunConfiguration
    | dict[str, Any]
    | None = None,
    json_configurations: dict[str, Any] | None = None,
    input_dir_path: str | None = 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 nextmv.ContentFormat.JSON, then the input data is extracted from the .data property. If you want to work with nextmv.ContentFormat.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. 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.ContentFormat.MULTI_FILE When working with JSON data, use the input argument directly. TYPE: Input | dict[str, Any] | BaseModel | str DEFAULT: None
name	Name of the local run. TYPE: str | None DEFAULT: None
description	Description of the local run. TYPE: str | None 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: Options | dict[str, str] | None 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: RunConfiguration | dict[str, Any] | None DEFAULT: None
json_configurations	Optional configurations for JSON serialization. This is used to customize the serialization before data is sent. TYPE: dict[str, Any] | None DEFAULT: None
input_dir_path	Path to a directory containing input files. This is useful for input formats like nextmv.ContentFormat.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: str | None 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 app.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: Input | dict[str, Any] | BaseModel | str = None,
    name: str | None = None,
    description: str | None = None,
    options: Options | dict[str, str] | None = None,
    configuration: RunConfiguration | dict[str, Any] | None = None,
    json_configurations: dict[str, Any] | None = None,
    input_dir_path: str | None = 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
        `nextmv.ContentFormat.JSON`, then the input data is extracted from
        the `.data` property.

        If you want to work with `nextmv.ContentFormat.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. 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.ContentFormat.MULTI_FILE`

        When working with JSON data, use the `input` argument directly.
    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.ContentFormat.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 app.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}")
    """

    configuration = 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")

    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.app_id if self.app_id is not None else "",
        src=self.src,
        manifest=self.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: Input | dict[str, Any] | BaseModel | str = None,
    name: str | None = None,
    description: str | None = None,
    run_options: Options | dict[str, str] | None = None,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
    configuration: RunConfiguration
    | dict[str, Any]
    | None = None,
    json_configurations: dict[str, Any] | None = None,
    input_dir_path: str | None = None,
    output_dir_path: str | None = ".",
) -> 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 nextmv.ContentFormat.JSON, then the input data is extracted from the .data property. If you want to work with nextmv.ContentFormat.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. 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.ContentFormat.MULTI_FILE When working with JSON data, use the input argument directly. TYPE: Input | dict[str, Any] | BaseModel | str DEFAULT: None
name	Name of the local run. TYPE: str | None DEFAULT: None
description	Description of the local run. TYPE: str | None 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: Options | dict[str, str] | None 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: RunConfiguration | dict[str, Any] | None DEFAULT: None
json_configurations	Optional configurations for JSON serialization. This is used to customize the serialization before data is sent. TYPE: dict[str, Any] | None DEFAULT: None
input_dir_path	Path to a directory containing input files. This is useful for input formats like nextmv.ContentFormat.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: str | None 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 app.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: Input | dict[str, Any] | BaseModel | str = None,
    name: str | None = None,
    description: str | None = None,
    run_options: Options | dict[str, str] | None = None,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
    configuration: RunConfiguration | dict[str, Any] | None = None,
    json_configurations: dict[str, Any] | None = None,
    input_dir_path: str | None = None,
    output_dir_path: str | None = ".",
) -> 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
        `nextmv.ContentFormat.JSON`, then the input data is extracted from
        the `.data` property.

        If you want to work with `nextmv.ContentFormat.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. 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.ContentFormat.MULTI_FILE`

        When working with JSON data, use the `input` argument directly.
    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.ContentFormat.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 app.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,
    )

register

register() -> None

Register the local application in the local registry.

This method adds an entry for the application in the local registry, which is stored as a YAML file at $HOME/.nextmv/registry.yaml. The registry keeps track of locally registered applications, allowing you to manage and list them easily.

If the application is already registered, this method does nothing.

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

def register(self) -> None:
    """
    Register the local application in the local registry.

    This method adds an entry for the application in the local registry,
    which is stored as a YAML file at `$HOME/.nextmv/registry.yaml`. The
    registry keeps track of locally registered applications, allowing you
    to manage and list them easily.

    If the application is already registered, this method does nothing.
    """

    reg = Registry.from_yaml()
    entry = reg.register(src=self.src, app_id=self.app_id, description=self.description)
    self.app_id = entry.app_id
    self.src = entry.src

run_input

run_input(
    run_id: str, output_dir_path: str | None = "."
) -> dict[str, Any] | str | None

Get the input of a local run.

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

PARAMETER	DESCRIPTION
run_id	ID of the run to retrieve input for. TYPE: str
output_dir_path	Path to a directory where non-JSON input files will be saved. This is required if the input 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
dict[str, Any] | str | None	The input of the run depending on the input format. If the input format is JSON, a dict is returned. If the input format is text, a string is returned. For multi-file and csv-archive, the input files are saved to the given output_dir_path, and the method returns None.

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

def run_input(self, run_id: str, output_dir_path: str | None = ".") -> dict[str, Any] | str | None:
    """
    Get the input of a local run.

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

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

    Returns
    -------
    dict[str, Any] | str | None
        The input of the run depending on the input format. If the input
        format is JSON, a dict is returned. If the input format is text, a
        string is returned. For multi-file and csv-archive, the input files
        are saved to the given `output_dir_path`, and the method returns
        `None`.
    """

    runs_dir = os.path.join(self.src, NEXTMV_DIR, RUNS_KEY)
    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}")

    # See whether we can attach the output directly or need to save to the given
    # directory
    run_information = self.run_metadata(run_id=run_id)
    input_type = run_information.metadata.format.format_input.input_type
    if input_type != ContentFormat.JSON and (not output_dir_path or output_dir_path == ""):
        raise ValueError(
            "The format of the input is not JSON: an `output_dir_path` must be provided.",
        )

    inputs_path = os.path.join(run_dir, INPUTS_KEY)
    if input_type == ContentFormat.JSON:
        input_file = os.path.join(inputs_path, DEFAULT_INPUT_JSON_FILE)
        if not os.path.exists(input_file):
            return None

        with open(input_file) as f:
            input_data = json.load(f)

        return input_data

    elif input_type == InputFormat.TEXT:
        input_file = os.path.join(inputs_path, DEFAULT_INPUT_TEXT_FILE)
        if not os.path.exists(input_file):
            return None

        with open(input_file) as f:
            input_data = f.read()

        return input_data

    elif input_type in {InputFormat.CSV_ARCHIVE, ContentFormat.MULTI_FILE}:
        if not os.path.exists(inputs_path):
            return None

        output_dir = os.path.join(output_dir_path)
        os.makedirs(output_dir, exist_ok=True)
        shutil.copytree(inputs_path, output_dir, dirs_exist_ok=True)

        return None

    raise ValueError(f"Unsupported input format: {input_type}")

run_logs

run_logs(run_id: str) -> str

Get the logs of a local run.

If the run does not have any logs, or they are empty, then this method simply returns a blank string. This method is equivalent to fetching the content of the .nextmv/runs/{run_id}/logs/logs.log file.

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

RETURNS	DESCRIPTION
str	The contents of the logs file for the run.

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_logs(self, run_id: str) -> str:
    """
    Get the logs of a local run.

    If the run does not have any logs, or they are empty, then this method
    simply returns a blank string. This method is equivalent to fetching
    the content of the `.nextmv/runs/{run_id}/logs/logs.log` file.

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

    Returns
    -------
    str
        The contents of the logs file for the run.

    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_DIR, RUNS_KEY)
    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}")

    logs_dir = os.path.join(run_dir, LOGS_KEY)
    if not os.path.exists(logs_dir):
        return ""

    logs_file = os.path.join(logs_dir, LOGS_FILE)
    if not os.path.exists(logs_file):
        return ""

    with open(logs_file) as f:
        logs = f.read()

    return logs

run_logs_with_polling

run_logs_with_polling(
    run_id: str,
    verbose: bool = False,
    rich_print: bool = False,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
    log_func: Callable[[str], None] | None = None,
) -> list[str]

Get the logs of a local run with polling.

Retrieves the logs of a run. This method polls for the logs until the run finishes executing or the polling strategy is exhausted. It is the "real-time" equivalent of the run_logs method. After the polling is done, all the logs are returned as a list of strings (one per line). You can use the verbose parameter to print the logs as they are obtained during the polling process. You can also use the rich_print parameter to enable rich printing for better formatting of the logs.

PARAMETER	DESCRIPTION
run_id	ID of the run to retrieve the logs for. TYPE: str
verbose	Whether to print the logs as they are obtained during the polling process. TYPE: bool DEFAULT: False
rich_print	Whether to use rich printing for better formatting of the logs. TYPE: bool DEFAULT: False
polling_options	Options to use when polling for the run logs. TYPE: PollingOptions DEFAULT: DEFAULT_POLLING_OPTIONS
log_func	Optional custom logging function callback. This function is invoked independently of the verbose flag. It needs to take a str as its argument and return None. TYPE: Optional[Callable[[str], None]] DEFAULT: None

RETURNS	DESCRIPTION
list[str]	List of log lines of the run.

RAISES	DESCRIPTION
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 import PollingOptions
>>> polling_opts = PollingOptions(max_tries=50, max_duration=600)
>>> logs = app.run_logs_with_polling("run-123", polling_opts)
>>> for line in logs:
...     print(line)
Starting optimization...
Found initial solution

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

def run_logs_with_polling(
    self,
    run_id: str,
    verbose: bool = False,
    rich_print: bool = False,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
    log_func: Callable[[str], None] | None = None,
) -> list[str]:
    """
    Get the logs of a local run with polling.

    Retrieves the logs of a run. This method polls for the logs until the
    run finishes executing or the polling strategy is exhausted. It is the
    "real-time" equivalent of the `run_logs` method. After the polling is
    done, all the logs are returned as a list of strings (one per line).
    You can use the `verbose` parameter to print the logs as they are
    obtained during the polling process. You can also use the `rich_print`
    parameter to enable rich printing for better formatting of the logs.

    Parameters
    ----------
    run_id : str
        ID of the run to retrieve the logs for.
    verbose : bool, default=False
        Whether to print the logs as they are obtained during the polling
        process.
    rich_print : bool, default=False
        Whether to use rich printing for better formatting of the logs.
    polling_options : PollingOptions, default=DEFAULT_POLLING_OPTIONS
        Options to use when polling for the run logs.
    log_func : Optional[Callable[[str], None]], default=None
        Optional custom logging function callback. This function is invoked
        independently of the `verbose` flag. It needs to take a `str` as
        its argument and return `None`.

    Returns
    -------
    list[str]
        List of log lines of the run.

    Raises
    ------
    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 import PollingOptions
    >>> polling_opts = PollingOptions(max_tries=50, max_duration=600)
    >>> logs = app.run_logs_with_polling("run-123", polling_opts)
    >>> for line in logs:
    ...     print(line)
    Starting optimization...
    Found initial solution
    """

    all_logs: list[str] = []
    lines_seen = 0

    def polling_func() -> tuple[Any, bool]:
        nonlocal lines_seen

        # Check if the run has finalized. If so, all logs are guaranteed
        # to have been flushed to the file before this check returned.
        run_information = self.run_metadata(run_id=run_id)
        is_done = run_information.metadata.run_is_finalized()

        # Read the current log file contents and emit only new lines.
        log_contents = self.run_logs(run_id)
        current_lines = log_contents.splitlines() if log_contents else []
        for line in current_lines[lines_seen:]:
            if log_func is not None:
                log_func(line)
            if verbose:
                if rich_print:
                    rich.print(line, file=sys.stderr)
                else:
                    log(line)
            all_logs.append(line)
        lines_seen = len(current_lines)

        return all_logs, is_done

    all_logs = poll(polling_options=polling_options, polling_func=polling_func)

    return all_logs

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_DIR, RUNS_KEY)
    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: str | None = "."
) -> 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: str | None = ".") -> 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: str | None = ".",
) -> 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 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: str | None = ".",
) -> 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 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.run_is_finalized():
            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) -> list[str]

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

RETURNS	DESCRIPTION
list[str]	The local URLs of the visual files that were opened in the web browser.

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) -> list[str]:
    """
    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.

    Returns
    -------
    list[str]
        The local URLs of the visual files that were opened in the web browser.

    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_DIR, RUNS_KEY)
    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}")

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

    return urls

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: list[str] | None = None,
    instance_id: str | None = None,
    verbose: bool | None = False,
    rich_print: bool | None = 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
rich_print	Whether to use rich printing for output during the sync process. 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: list[str] | None = None,
    instance_id: str | None = None,
    verbose: bool | None = False,
    rich_print: bool | None = 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.
    rich_print : Optional[bool], default=False
        Whether to use rich printing for output during the sync process.

    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 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:
        if rich_print:
            rich.print(
                f":cloud_with_lightning:  Starting sync of local application [magenta]{self.src}[/magenta] to "
                f"Nextmv Cloud application [magenta]{target.id}[/magenta].",
                file=sys.stderr,
            )
        else:
            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_DIR, RUNS_KEY)
        if not run_ids:
            # If runs are not specified, by default we sync all local runs that
            # can be found.
            if not os.path.exists(runs_dir):
                raise ValueError(
                    f"`.nextmv/runs` dir does not exist at app source: {self.src}, try making a run first"
                )

            dirs = os.listdir(runs_dir)
            run_ids = [d for d in dirs if os.path.isdir(os.path.join(runs_dir, d))]

            if verbose:
                if rich_print:
                    rich.print(
                        f":bulb: Found [magenta]{len(run_ids)}[/magenta] local runs to "
                        f"sync from [magenta]{runs_dir}[/magenta].",
                        file=sys.stderr,
                    )
                else:
                    log(f"ℹ️  Found {len(run_ids)} local runs to sync from {runs_dir}.")
        else:
            if verbose:
                if rich_print:
                    rich.print(
                        f":bulb: Syncing [magenta]{len(run_ids)}[/magenta] specified local runs from "
                        f"[magenta]{runs_dir}[/magenta].",
                        file=sys.stderr,
                    )
                else:
                    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,
                rich_print=rich_print,
            )
            if synced:
                total += 1

        if not verbose:
            return

        if rich_print:
            rich.print(
                f":rocket: Process completed, synced local application [magenta]{self.src}[/magenta] to "
                f"Nextmv Cloud application [magenta]{target.id}[/magenta]: "
                f"[magenta]{total}[/magenta]/[magenta]{len(run_ids)}[/magenta] runs.",
                file=sys.stderr,
            )
            return

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

update

update(
    app_id: str | None = None,
    description: str | None = None,
    content_format: ContentFormat | None = None,
) -> None

Update the local application.

This method allows you to update the local application's metadata, such as its app_id, description and content format. It does not affect the application's source code or runs.

PARAMETER	DESCRIPTION
app_id	New app_id for the application. If None, the app_id is not updated. TYPE: Optional[str] DEFAULT: None
description	New description for the application. If None, the description is not updated. TYPE: Optional[str] DEFAULT: None
content_format	New content format for the application. If None, the content format is not updated. TYPE: Optional[ContentFormat] DEFAULT: None

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

def update(
    self,
    app_id: str | None = None,
    description: str | None = None,
    content_format: ContentFormat | None = None,
) -> None:
    """
    Update the local application.

    This method allows you to update the local application's metadata, such
    as its app_id, description and content format. It does not affect
    the application's source code or runs.

    Parameters
    ----------
    app_id : Optional[str]
        New app_id for the application. If None, the app_id is not updated.
    description : Optional[str]
        New description for the application. If None, the description is not
        updated.
    content_format : Optional[ContentFormat]
        New content format for the application. If None, the content format
        is not updated.
    """

    reg = Registry.from_yaml()

    # Use both app_id and src to identify the entry for maximum specificity
    updated_entry = reg.update_entry(
        src=self.src,
        app_id=self.app_id,
        new_app_id=app_id,
        description=description,
        content_format=content_format,
    )

    if updated_entry is None:
        raise ValueError(f"Application with app_id `{self.app_id}` and src `{self.src}` is not registered locally.")

    # Update the instance attributes to reflect the changes
    if app_id is not None:
        self.app_id = app_id
    if description is not None:
        self.description = description
    if content_format is not None:
        self.content_format = content_format