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

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: InputFormat | None = None

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

    try:
        manifest = 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,
        manifest=manifest,
        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,
) -> 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: str | None = None,
    description: str | None = None,
    destination: str | None = 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,
    )

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 = InputFormat.JSON
    if self.manifest.configuration is not None and self.manifest.configuration.content is not None:
        content_format = self.manifest.configuration.content.format

    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 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. 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. 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.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: `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 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. 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.
    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 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 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. 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. 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.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: `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 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. 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.
    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 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 != InputFormat.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 == InputFormat.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, InputFormat.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_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.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: 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.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.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: InputFormat | 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[InputFormat]` DEFAULT: `None`

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

def update(
    self,
    app_id: str | None = None,
    description: str | None = None,
    content_format: InputFormat | 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[InputFormat]
        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

Application Module¶

application ¶

Application ¶

src ¶

description ¶

app_id class-attribute instance-attribute ¶

content_format class-attribute instance-attribute ¶

description class-attribute instance-attribute ¶

from_registry classmethod ¶

src ¶

app_id ¶

get_or_register classmethod ¶

app_src ¶

app_id ¶

initialize classmethod ¶

src ¶

description ¶

destination ¶

is_registered ¶

list_runs ¶

status ¶

manifest class-attribute instance-attribute ¶

model_post_init ¶

new_run ¶

input ¶

name ¶

description ¶

options ¶

configuration ¶

json_configurations ¶

input_dir_path ¶

new_run_with_result ¶

input ¶

name ¶

description ¶

run_options ¶

polling_options ¶

configuration ¶

json_configurations ¶

input_dir_path ¶

output_dir_path ¶

register ¶

run_input ¶

run_id ¶

output_dir_path ¶

run_logs ¶

run_id ¶

run_metadata ¶

run_id ¶

run_result ¶

run_id ¶

output_dir_path ¶

run_result_with_polling ¶

run_id ¶

polling_options ¶

output_dir_path ¶

run_visuals ¶

run_id ¶

src instance-attribute ¶

sync ¶

target ¶

run_ids ¶

instance_id ¶

verbose ¶

rich_print ¶

update ¶

app_id ¶

description ¶

content_format ¶

`src` ¶

`description` ¶

app_id `class-attribute` `instance-attribute` ¶

content_format `class-attribute` `instance-attribute` ¶

description `class-attribute` `instance-attribute` ¶

from_registry `classmethod` ¶

`src` ¶

`app_id` ¶

get_or_register `classmethod` ¶

`app_src` ¶

`app_id` ¶

initialize `classmethod` ¶

`src` ¶

`description` ¶

`destination` ¶

`status` ¶

manifest `class-attribute` `instance-attribute` ¶

`input` ¶

`name` ¶

`description` ¶

`options` ¶

`configuration` ¶

`json_configurations` ¶

`input_dir_path` ¶

`input` ¶

`name` ¶

`description` ¶

`run_options` ¶

`polling_options` ¶

`configuration` ¶

`json_configurations` ¶

`input_dir_path` ¶

`output_dir_path` ¶

`run_id` ¶

`output_dir_path` ¶

`run_id` ¶

`run_id` ¶

`run_id` ¶

`output_dir_path` ¶

`run_id` ¶

`polling_options` ¶

`output_dir_path` ¶

`run_id` ¶

src `instance-attribute` ¶

`target` ¶

`run_ids` ¶

`instance_id` ¶

`verbose` ¶

`rich_print` ¶

`app_id` ¶

`description` ¶

`content_format` ¶