Application Module

This section documents the application components of the Nextmv Cloud API.

The application module is organized with the main Application class and several mixin classes that provide specialized functionality. Every method belonging to a mixin class can be accessed through an instance of the Application.

Main Class

application

Application module for interacting with Nextmv Cloud applications.

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

CLASS	DESCRIPTION
ApplicationType	Enumeration of application types in Nextmv Cloud.
Application	Class for interacting with applications in Nextmv Cloud.

FUNCTION	DESCRIPTION
list_application	Function to list applications in Nextmv Cloud.

Application

Bases: BaseModel, ApplicationAcceptanceMixin, ApplicationBatchMixin, ApplicationRunMixin, ApplicationEnsembleMixin, ApplicationInstanceMixin, ApplicationSecretsMixin, ApplicationVersionMixin, ApplicationInputSetMixin, ApplicationManagedInputMixin, ApplicationShadowMixin, ApplicationSwitchbackMixin

A published decision model that can be executed.

You can import the Application class directly from cloud:

from nextmv.cloud import Application

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

Note: It is recommended to use Application.get() or Application.new() instead of direct initialization to ensure proper setup.

ATTRIBUTE	DESCRIPTION
id	ID of the application. TYPE: str
name	Name of the application. TYPE: (str, optional)
description	Description of the application. TYPE: (str, optional)
type	Type of the application (CUSTOM, SUBSCRIPTION, or PIPELINE). TYPE: (ApplicationType, optional)
default_instance_id	Default instance ID to use for submitting runs. TYPE: (str, optional)
default_experiment_instance	Default experiment instance ID to use for experiments. TYPE: (str, optional)
subscription_id	Subscription ID if the application is a subscription type. TYPE: (str, optional)
locked	Whether the application is locked. TYPE: bool, default=False
created_at	Creation timestamp of the application. TYPE: (datetime, optional)
updated_at	Last update timestamp of the application. TYPE: (datetime, optional)
client	Client to use for interacting with the Nextmv Cloud API. This is an SDK-specific attribute and it is not part of the API representation of an application. TYPE: Client
endpoint	Base endpoint for the application. This is an SDK-specific attribute and it is not part of the API representation of an application. TYPE: str, default="v1/applications/{id}"
experiments_endpoint	Base endpoint for experiments. This is an SDK-specific attribute and it is not part of the API representation of an application. TYPE: str, default="{base}/experiments"
ensembles_endpoint	Base endpoint for ensembles. This is an SDK-specific attribute and it is not part of the API representation of an application. TYPE: str, default="{base}/ensembles"

Examples:

>>> from nextmv.cloud import Client, Application
>>> client = Client(api_key="your-api-key")
>>> # Retrieve an existing application
>>> app = Application.get(client=client, id="your-app-id")
>>> print(f"Application name: {app.name}")
Application name: My Application
>>> # Create a new application
>>> new_app = Application.new(client=client, name="My New App", id="my-new-app")
>>> # List application instances
>>> instances = app.list_instances()

Returned by:

• nextmv ☁️ Cloud Reference Application Module  application
  •  Application
    •  get
    •  new
    •  update
  •  list_applications

client class-attribute instance-attribute

client: Client = Field(exclude=True)

Client to use for interacting with the Nextmv Cloud API.

created_at class-attribute instance-attribute

created_at: datetime | None = None

Creation timestamp of the application.

default_experiment_instance class-attribute instance-attribute

default_experiment_instance: str | None = None

Default experiment instance ID to use for experiments.

default_instance_id class-attribute instance-attribute

default_instance_id: str | None = Field(
    serialization_alias="default_instance",
    validation_alias=AliasChoices(
        "default_instance", "default_instance_id"
    ),
    default=None,
)

Default instance ID to use for submitting runs.

delete

delete() -> None

Delete the application.

Permanently removes the application from Nextmv Cloud.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> app.delete()  # Permanently deletes the application

Source code in nextmv/nextmv/cloud/application/__init__.py

def delete(self) -> None:
    """
    Delete the application.

    Permanently removes the application from Nextmv Cloud.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> app.delete()  # Permanently deletes the application
    """

    _ = self.client.request(
        method="DELETE",
        endpoint=self.endpoint,
    )

description class-attribute instance-attribute

description: str | None = None

Description of the application.

endpoint class-attribute instance-attribute

endpoint: str = Field(
    exclude=True, default="v1/applications/{id}"
)

Base endpoint for the application.

ensembles_endpoint class-attribute instance-attribute

ensembles_endpoint: str = Field(
    exclude=True, default="{base}/ensembles"
)

Base endpoint for managing the ensemble definitions in the application

exists staticmethod

exists(client: Client, id: str) -> bool

Check if an application exists.

PARAMETER	DESCRIPTION
client	Client to use for interacting with the Nextmv Cloud API. TYPE: Client
id	ID of the application to check. TYPE: str

RETURNS	DESCRIPTION
bool	True if the application exists, False otherwise.

Examples:

>>> from nextmv.cloud import Client
>>> client = Client(api_key="your-api-key")
>>> Application.exists(client, "app-123")
True

Source code in nextmv/nextmv/cloud/application/__init__.py

@staticmethod
def exists(client: Client, id: str) -> bool:
    """
    Check if an application exists.

    Parameters
    ----------
    client : Client
        Client to use for interacting with the Nextmv Cloud API.
    id : str
        ID of the application to check.

    Returns
    -------
    bool
        True if the application exists, False otherwise.

    Examples
    --------
    >>> from nextmv.cloud import Client
    >>> client = Client(api_key="your-api-key")
    >>> Application.exists(client, "app-123")
    True
    """

    try:
        _ = client.request(
            method="GET",
            endpoint=f"v1/applications/{id}",
        )
        # If the request was successful, the application exists.
        return True
    except requests.HTTPError as e:
        if _is_not_exist_error(e):
            return False
        # Re-throw the exception if it is not the expected 404 error.
        raise e from None

experiments_endpoint class-attribute instance-attribute

experiments_endpoint: str = Field(
    exclude=True, default="{base}/experiments"
)

Base endpoint for the experiments in the application.

get classmethod

get(client: Client, id: str) -> Application

Retrieve an application directly from Nextmv Cloud.

This function is useful if you want to populate an Application class by fetching the attributes directly from Nextmv Cloud.

PARAMETER	DESCRIPTION
client	Client to use for interacting with the Nextmv Cloud API. TYPE: Client
id	ID of the application to retrieve. TYPE: str

RETURNS	DESCRIPTION
Application	The requested application.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/__init__.py

@classmethod
def get(cls, client: Client, id: str) -> "Application":
    """
    Retrieve an application directly from Nextmv Cloud.

    This function is useful if you want to populate an `Application` class
    by fetching the attributes directly from Nextmv Cloud.

    Parameters
    ----------
    client : Client
        Client to use for interacting with the Nextmv Cloud API.
    id : str
        ID of the application to retrieve.

    Returns
    -------
    Application
        The requested application.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    response = client.request(
        method="GET",
        endpoint=f"v1/applications/{id}",
    )

    return cls.from_dict({"client": client} | response.json())

id instance-attribute

id: str

ID of the application.

locked class-attribute instance-attribute

locked: bool = False

Whether the application is locked.

model_post_init

model_post_init(__context) -> None

Initialize the endpoint and experiments_endpoint attributes.

This method is automatically called after class initialization to format the endpoint and experiments_endpoint URLs with the application ID.

Source code in nextmv/nextmv/cloud/application/__init__.py

def model_post_init(self, __context) -> None:
    """Initialize the endpoint and experiments_endpoint attributes.

    This method is automatically called after class initialization to
    format the endpoint and experiments_endpoint URLs with the application ID.
    """
    self.endpoint = self.endpoint.format(id=self.id)
    self.experiments_endpoint = self.experiments_endpoint.format(base=self.endpoint)
    self.ensembles_endpoint = self.ensembles_endpoint.format(base=self.endpoint)

name class-attribute instance-attribute

name: str | None = None

Name of the application.

new classmethod

new(
    client: Client,
    name: str | None = None,
    id: str | None = None,
    description: str | None = None,
    is_workflow: bool | None = None,
    exist_ok: bool = False,
    default_instance_id: str | None = None,
    default_experiment_instance: str | None = None,
) -> Application

Create a new application directly in Nextmv Cloud.

The application is created as an empty shell, and executable code must be pushed to the app before running it remotely.

PARAMETER	DESCRIPTION
client	Client to use for interacting with the Nextmv Cloud API. TYPE: Client
name	Name of the application. Uses the ID as the name if not provided. TYPE: str | None = None DEFAULT: None
id	ID of the application. Will be generated if not provided. TYPE: str | None = None DEFAULT: None
description	Description of the application. TYPE: str | None = None DEFAULT: None
is_workflow	Whether the application is a Decision Workflow. TYPE: bool | None = None DEFAULT: None
exist_ok	If True and an application with the same ID already exists, return the existing application instead of creating a new one. TYPE: bool DEFAULT: False
default_instance_id	Default instance ID to use for submitting runs. TYPE: str DEFAULT: None
default_experiment_instance	Default experiment instance ID to use for experiments. TYPE: str DEFAULT: None

RETURNS	DESCRIPTION
Application	The newly created (or existing) application.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> from nextmv.cloud import Client
>>> client = Client(api_key="your-api-key")
>>> app = Application.new(client=client, name="My New App", id="my-app")

Source code in nextmv/nextmv/cloud/application/__init__.py

@classmethod
def new(
    cls,
    client: Client,
    name: str | None = None,
    id: str | None = None,
    description: str | None = None,
    is_workflow: bool | None = None,
    exist_ok: bool = False,
    default_instance_id: str | None = None,
    default_experiment_instance: str | None = None,
) -> "Application":
    """
    Create a new application directly in Nextmv Cloud.

    The application is created as an empty shell, and executable code must
    be pushed to the app before running it remotely.

    Parameters
    ----------
    client : Client
        Client to use for interacting with the Nextmv Cloud API.
    name : str | None = None
        Name of the application. Uses the ID as the name if not provided.
    id : str | None = None
        ID of the application. Will be generated if not provided.
    description : str | None = None
        Description of the application.
    is_workflow : bool | None = None
        Whether the application is a Decision Workflow.
    exist_ok : bool, default=False
        If True and an application with the same ID already exists,
        return the existing application instead of creating a new one.
    default_instance_id : str, optional
        Default instance ID to use for submitting runs.
    default_experiment_instance : str, optional
        Default experiment instance ID to use for experiments.

    Returns
    -------
    Application
        The newly created (or existing) application.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> from nextmv.cloud import Client
    >>> client = Client(api_key="your-api-key")
    >>> app = Application.new(client=client, name="My New App", id="my-app")
    """

    if exist_ok and (id is None or id == ""):
        raise ValueError("If exist_ok is True, id must be provided")

    if id is None or id == "":
        id = safe_id("app")

    if exist_ok and cls.exists(client=client, id=id):
        response = client.request(
            method="GET",
            endpoint=f"v1/applications/{id}",
        )

        return cls.from_dict({"client": client} | response.json())

    if name is None or name == "":
        name = id

    payload = {
        "name": name,
        "id": id,
    }

    if description is not None:
        payload["description"] = description

    if is_workflow is not None:
        payload["is_pipeline"] = is_workflow

    if default_instance_id is not None:
        payload["default_instance"] = default_instance_id

    if default_experiment_instance is not None:
        payload["default_experiment_instance"] = default_experiment_instance

    response = client.request(
        method="POST",
        endpoint="v1/applications",
        payload=payload,
    )

    return cls.from_dict({"client": client} | response.json())

push

push(
    manifest: Manifest | None = None,
    app_dir: str | None = None,
    verbose: bool = False,
    model: Model | None = None,
    model_configuration: ModelConfiguration | None = None,
    rich_print: bool = False,
) -> None

Push an app to Nextmv Cloud.

If the manifest is not provided, an app.yaml file will be searched for in the provided path. If there is no manifest file found, an exception will be raised.

There are two ways to push an app to Nextmv Cloud: 1. Specifying app_dir, which is the path to an app's root directory. This acts as an external strategy, where the app is composed of files in a directory and those apps are packaged and pushed to Nextmv Cloud. 2. Specifying a model and model_configuration. This acts as an internal (or Python-native) strategy, where the app is actually a nextmv.Model. The model is encoded, some dependencies and accompanying files are packaged, and the app is pushed to Nextmv Cloud.

PARAMETER	DESCRIPTION
manifest	The manifest for the app. If None, an app.yaml file in the provided app directory will be used. TYPE: Optional[Manifest] DEFAULT: None
app_dir	The path to the app's root directory. If None, the current directory will be used. This is for the external strategy approach. TYPE: Optional[str] DEFAULT: None
verbose	Whether to print verbose output during the push process. TYPE: bool DEFAULT: False
model	The Python-native model to push. Must be specified together with model_configuration. This is for the internal strategy approach. TYPE: Optional[Model] DEFAULT: None
model_configuration	Configuration for the Python-native model. Must be specified together with model. TYPE: Optional[ModelConfiguration] DEFAULT: None
rich_print	Whether to use rich printing when verbose output is enabled. TYPE: bool DEFAULT: False

RAISES	DESCRIPTION
ValueError	If neither app_dir nor model/model_configuration is provided correctly, or if only one of model and model_configuration is provided.
TypeError	If model is not an instance of nextmv.Model or if model_configuration is not an instance of nextmv.ModelConfiguration.
Exception	If there's an error in the build, packaging, or cleanup process.

Examples:

1. Push an app using an external strategy (directory-based):

>>> import os
>>> from nextmv import cloud
>>> client = cloud.Client(api_key=os.getenv("NEXTMV_API_KEY"))
>>> app = cloud.Application(client=client, id="<YOUR-APP-ID>")
>>> app.push()  # Use verbose=True for step-by-step output.

1. Push an app using an internal strategy (Python-native model):

>>> import os
>>> import nextroute
>>> import nextmv
>>> import nextmv.cloud
>>>
>>> # Define the model that makes decisions
>>> class DecisionModel(nextmv.Model):
...     def solve(self, input: nextmv.Input) -> nextmv.Output:
...         nextroute_input = nextroute.schema.Input.from_dict(input.data)
...         nextroute_options = nextroute.Options.extract_from_dict(input.options.to_dict())
...         nextroute_output = nextroute.solve(nextroute_input, nextroute_options)
...
...         return nextmv.Output(
...             options=input.options,
...             solution=nextroute_output.solutions[0].to_dict(),
...             statistics=nextroute_output.statistics.to_dict(),
...         )
>>>
>>> # Define the options that the model needs
>>> opt = []
>>> default_options = nextroute.Options()
>>> for name, default_value in default_options.to_dict().items():
...     opt.append(nextmv.Option(name.lower(), type(default_value), default_value, name, False))
>>> options = nextmv.Options(*opt)
>>>
>>> # Instantiate the model and model configuration
>>> model = DecisionModel()
>>> model_configuration = nextmv.ModelConfiguration(
...     name="python_nextroute_model",
...     requirements=[
...         "nextroute==1.8.1",
...         "nextmv==0.14.0.dev1",
...     ],
...     options=options,
... )
>>>
>>> # Push the model to Nextmv Cloud
>>> client = cloud.Client(api_key=os.getenv("NEXTMV_API_KEY"))
>>> app = cloud.Application(client=client, id="<YOUR-APP-ID>")
>>> manifest = nextmv.cloud.default_python_manifest()
>>> app.push(
...     manifest=manifest,
...     verbose=True,
...     model=model,
...     model_configuration=model_configuration,
... )

Source code in nextmv/nextmv/cloud/application/__init__.py

def push(  # noqa: C901
    self,
    manifest: Manifest | None = None,
    app_dir: str | None = None,
    verbose: bool = False,
    model: Model | None = None,
    model_configuration: ModelConfiguration | None = None,
    rich_print: bool = False,
) -> None:
    """
    Push an app to Nextmv Cloud.

    If the manifest is not provided, an `app.yaml` file will be searched
    for in the provided path. If there is no manifest file found, an
    exception will be raised.

    There are two ways to push an app to Nextmv Cloud:
    1. Specifying `app_dir`, which is the path to an app's root directory.
    This acts as an external strategy, where the app is composed of files
    in a directory and those apps are packaged and pushed to Nextmv Cloud.
    2. Specifying a `model` and `model_configuration`. This acts as an
    internal (or Python-native) strategy, where the app is actually a
    `nextmv.Model`. The model is encoded, some dependencies and
    accompanying files are packaged, and the app is pushed to Nextmv Cloud.

    Parameters
    ----------
    manifest : Optional[Manifest], default=None
        The manifest for the app. If None, an `app.yaml` file in the provided
        app directory will be used.
    app_dir : Optional[str], default=None
        The path to the app's root directory. If None, the current directory
        will be used. This is for the external strategy approach.
    verbose : bool, default=False
        Whether to print verbose output during the push process.
    model : Optional[Model], default=None
        The Python-native model to push. Must be specified together with
        `model_configuration`. This is for the internal strategy approach.
    model_configuration : Optional[ModelConfiguration], default=None
        Configuration for the Python-native model. Must be specified together
        with `model`.
    rich_print : bool, default=False
        Whether to use rich printing when verbose output is enabled.

    Raises
    ------
    ValueError
        If neither app_dir nor model/model_configuration is provided correctly,
        or if only one of model and model_configuration is provided.
    TypeError
        If model is not an instance of nextmv.Model or if model_configuration
        is not an instance of nextmv.ModelConfiguration.
    Exception
        If there's an error in the build, packaging, or cleanup process.

    Examples
    --------
    1. Push an app using an external strategy (directory-based):

    >>> import os
    >>> from nextmv import cloud
    >>> client = cloud.Client(api_key=os.getenv("NEXTMV_API_KEY"))
    >>> app = cloud.Application(client=client, id="<YOUR-APP-ID>")
    >>> app.push()  # Use verbose=True for step-by-step output.

    2. Push an app using an internal strategy (Python-native model):

    >>> import os
    >>> import nextroute
    >>> import nextmv
    >>> import nextmv.cloud
    >>>
    >>> # Define the model that makes decisions
    >>> class DecisionModel(nextmv.Model):
    ...     def solve(self, input: nextmv.Input) -> nextmv.Output:
    ...         nextroute_input = nextroute.schema.Input.from_dict(input.data)
    ...         nextroute_options = nextroute.Options.extract_from_dict(input.options.to_dict())
    ...         nextroute_output = nextroute.solve(nextroute_input, nextroute_options)
    ...
    ...         return nextmv.Output(
    ...             options=input.options,
    ...             solution=nextroute_output.solutions[0].to_dict(),
    ...             statistics=nextroute_output.statistics.to_dict(),
    ...         )
    >>>
    >>> # Define the options that the model needs
    >>> opt = []
    >>> default_options = nextroute.Options()
    >>> for name, default_value in default_options.to_dict().items():
    ...     opt.append(nextmv.Option(name.lower(), type(default_value), default_value, name, False))
    >>> options = nextmv.Options(*opt)
    >>>
    >>> # Instantiate the model and model configuration
    >>> model = DecisionModel()
    >>> model_configuration = nextmv.ModelConfiguration(
    ...     name="python_nextroute_model",
    ...     requirements=[
    ...         "nextroute==1.8.1",
    ...         "nextmv==0.14.0.dev1",
    ...     ],
    ...     options=options,
    ... )
    >>>
    >>> # Push the model to Nextmv Cloud
    >>> client = cloud.Client(api_key=os.getenv("NEXTMV_API_KEY"))
    >>> app = cloud.Application(client=client, id="<YOUR-APP-ID>")
    >>> manifest = nextmv.cloud.default_python_manifest()
    >>> app.push(
    ...     manifest=manifest,
    ...     verbose=True,
    ...     model=model,
    ...     model_configuration=model_configuration,
    ... )
    """

    if verbose:
        if rich_print:
            rich.print(f":cd: Starting build for Nextmv application [magenta]{self.id}[/magenta].", file=sys.stderr)
        else:
            log("💽 Starting build for Nextmv application.")

    if app_dir is None or app_dir == "":
        app_dir = "."

    if manifest is None:
        manifest = Manifest.from_yaml(app_dir)

    if model is not None and not isinstance(model, Model):
        raise TypeError("model must be an instance of nextmv.Model")

    if model_configuration is not None and not isinstance(model_configuration, ModelConfiguration):
        raise TypeError("model_configuration must be an instance of nextmv.ModelConfiguration")

    if (model is None and model_configuration is not None) or (model is not None and model_configuration is None):
        raise ValueError("model and model_configuration must be provided together")

    package._run_build_command(app_dir, manifest.build, verbose, rich_print)
    package._run_pre_push_command(app_dir, manifest.pre_push, verbose, rich_print)
    tar_file, output_dir = package._package(app_dir, manifest, model, model_configuration, verbose, rich_print)
    self.__update_app_binary(tar_file, manifest, verbose, rich_print)

    try:
        shutil.rmtree(output_dir)
    except OSError as e:
        raise Exception(f"error deleting output directory: {e}") from e

subscription_id class-attribute instance-attribute

subscription_id: str | None = None

Subscription ID if the application is a subscription type.

type class-attribute instance-attribute

type: ApplicationType | None = None

Type of the application.

update

update(
    name: str | None = None,
    description: str | None = None,
    default_instance_id: str | None = None,
    default_experiment_instance: str | None = None,
) -> Application

Update the application.

PARAMETER	DESCRIPTION
name	Optional name of the application. TYPE: Optional[str] DEFAULT: None
description	Optional description of the application. TYPE: Optional[str] DEFAULT: None
default_instance_id	Optional default instance ID for the application. TYPE: Optional[str] DEFAULT: None
default_experiment_instance	Optional default experiment instance ID for the application. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
Application	The updated application.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/__init__.py

def update(
    self,
    name: str | None = None,
    description: str | None = None,
    default_instance_id: str | None = None,
    default_experiment_instance: str | None = None,
) -> "Application":
    """
    Update the application.

    Parameters
    ----------
    name : Optional[str], default=None
        Optional name of the application.
    description : Optional[str], default=None
        Optional description of the application.
    default_instance_id : Optional[str], default=None
        Optional default instance ID for the application.
    default_experiment_instance : Optional[str], default=None
        Optional default experiment instance ID for the application.

    Returns
    -------
    Application
        The updated application.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    app = self.get(client=self.client, id=self.id)
    app_dict = app.to_dict()
    payload = app_dict.copy()

    if name is not None:
        payload["name"] = name
    if description is not None:
        payload["description"] = description
    if default_instance_id is not None:
        payload["default_instance"] = default_instance_id
    if default_experiment_instance is not None:
        payload["default_experiment_instance"] = default_experiment_instance

    response = self.client.request(
        method="PUT",
        endpoint=self.endpoint,
        payload=payload,
    )

    return Application.from_dict({"client": self.client} | response.json())

updated_at class-attribute instance-attribute

updated_at: datetime | None = None

Last update timestamp of the application.

upload_data

upload_data(
    upload_url: UploadURL | str,
    data: dict[str, Any] | str | None = None,
    json_configurations: dict[str, Any] | None = None,
    tar_file: str | None = None,
) -> None

Upload data to the provided upload URL.

This method allows uploading data (either a dictionary or string) to a pre-signed URL. If the data is a dictionary, it will be converted to a JSON string before upload.

PARAMETER	DESCRIPTION
upload_url	Upload URL object containing the pre-signed URL to use for uploading. If it is a string, it will be used directly as the pre-signed URL. TYPE: UploadURL | str
data	Data to upload. Can be either a dictionary that will be converted to JSON, or a pre-formatted JSON string. TYPE: Optional[Union[dict[str, Any], str]] DEFAULT: None
json_configurations	Optional configurations for JSON serialization. If provided, these configurations will be used when serializing the data via json.dumps. TYPE: Optional[dict[str, Any]] DEFAULT: None
tar_file	If provided, this will be used to upload a tar file instead of a JSON string or dictionary. This is useful for uploading large files that are already packaged as a tarball. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
None	This method doesn't return anything.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> # Upload a dictionary as JSON
>>> data = {"locations": [...], "vehicles": [...]}
>>> url = app.upload_url()
>>> app.upload_data(data=data, upload_url=url)
>>>
>>> # Upload a pre-formatted JSON string
>>> json_str = '{"locations": [...], "vehicles": [...]}'
>>> app.upload_data(data=json_str, upload_url=url)

Source code in nextmv/nextmv/cloud/application/__init__.py

def upload_data(
    self,
    upload_url: UploadURL | str,
    data: dict[str, Any] | str | None = None,
    json_configurations: dict[str, Any] | None = None,
    tar_file: str | None = None,
) -> None:
    """
    Upload data to the provided upload URL.

    This method allows uploading data (either a dictionary or string)
    to a pre-signed URL. If the data is a dictionary, it will be converted to
    a JSON string before upload.

    Parameters
    ----------
    upload_url : UploadURL | str
        Upload URL object containing the pre-signed URL to use for
        uploading. If it is a string, it will be used directly as the
        pre-signed URL.
    data : Optional[Union[dict[str, Any], str]]
        Data to upload. Can be either a dictionary that will be
        converted to JSON, or a pre-formatted JSON string.
    json_configurations : Optional[dict[str, Any]], default=None
        Optional configurations for JSON serialization. If provided, these
        configurations will be used when serializing the data via
        `json.dumps`.
    tar_file : Optional[str], default=None
        If provided, this will be used to upload a tar file instead of
        a JSON string or dictionary. This is useful for uploading large
        files that are already packaged as a tarball.

    Returns
    -------
    None
        This method doesn't return anything.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> # Upload a dictionary as JSON
    >>> data = {"locations": [...], "vehicles": [...]}
    >>> url = app.upload_url()
    >>> app.upload_data(data=data, upload_url=url)
    >>>
    >>> # Upload a pre-formatted JSON string
    >>> json_str = '{"locations": [...], "vehicles": [...]}'
    >>> app.upload_data(data=json_str, upload_url=url)
    """

    if data is not None and isinstance(data, dict):
        data = deflated_serialize_json(data, json_configurations=json_configurations)

    self.client.upload_to_presigned_url(
        url=upload_url.upload_url if isinstance(upload_url, UploadURL) else upload_url,
        data=data,
        tar_file=tar_file,
    )

upload_large_input

upload_large_input(
    input: dict[str, Any] | str | None,
    upload_url: UploadURL,
    json_configurations: dict[str, Any] | None = None,
    tar_file: str | None = None,
) -> None

Warning

upload_large_input is deprecated, use upload_data instead.

Upload large input data to the provided upload URL.

This method allows uploading large input data (either a dictionary or string) to a pre-signed URL. If the input is a dictionary, it will be converted to a JSON string before upload.

PARAMETER	DESCRIPTION
input	Input data to upload. Can be either a dictionary that will be converted to JSON, or a pre-formatted JSON string. TYPE: Optional[Union[dict[str, Any], str]]
upload_url	Upload URL object containing the pre-signed URL to use for uploading. TYPE: UploadURL
json_configurations	Optional configurations for JSON serialization. If provided, these configurations will be used when serializing the data via json.dumps. TYPE: Optional[dict[str, Any]] DEFAULT: None
tar_file	If provided, this will be used to upload a tar file instead of a JSON string or dictionary. This is useful for uploading large files that are already packaged as a tarball. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
None	This method doesn't return anything.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> # Upload a dictionary as JSON
>>> data = {"locations": [...], "vehicles": [...]}
>>> url = app.upload_url()
>>> app.upload_large_input(input=data, upload_url=url)
>>>
>>> # Upload a pre-formatted JSON string
>>> json_str = '{"locations": [...], "vehicles": [...]}'
>>> app.upload_large_input(input=json_str, upload_url=url)

Source code in nextmv/nextmv/cloud/application/__init__.py

def upload_large_input(
    self,
    input: dict[str, Any] | str | None,
    upload_url: UploadURL,
    json_configurations: dict[str, Any] | None = None,
    tar_file: str | None = None,
) -> None:
    """
    !!! warning
        `upload_large_input` is deprecated, use `upload_data` instead.

    Upload large input data to the provided upload URL.

    This method allows uploading large input data (either a dictionary or string)
    to a pre-signed URL. If the input is a dictionary, it will be converted to
    a JSON string before upload.

    Parameters
    ----------
    input : Optional[Union[dict[str, Any], str]]
        Input data to upload. Can be either a dictionary that will be
        converted to JSON, or a pre-formatted JSON string.
    upload_url : UploadURL
        Upload URL object containing the pre-signed URL to use for uploading.
    json_configurations : Optional[dict[str, Any]], default=None
        Optional configurations for JSON serialization. If provided, these
        configurations will be used when serializing the data via
        `json.dumps`.
    tar_file : Optional[str], default=None
        If provided, this will be used to upload a tar file instead of
        a JSON string or dictionary. This is useful for uploading large
        files that are already packaged as a tarball.

    Returns
    -------
    None
        This method doesn't return anything.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> # Upload a dictionary as JSON
    >>> data = {"locations": [...], "vehicles": [...]}
    >>> url = app.upload_url()
    >>> app.upload_large_input(input=data, upload_url=url)
    >>>
    >>> # Upload a pre-formatted JSON string
    >>> json_str = '{"locations": [...], "vehicles": [...]}'
    >>> app.upload_large_input(input=json_str, upload_url=url)
    """

    deprecated(
        name="Application.upload_large_input",
        reason="`upload_large_input` is deprecated, use `upload_data` instead",
    )

    self.upload_data(
        data=input,
        upload_url=upload_url,
        json_configurations=json_configurations,
        tar_file=tar_file,
    )

upload_url

upload_url() -> UploadURL

Get an upload URL to use for uploading a file.

This method generates a pre-signed URL that can be used to upload large files to Nextmv Cloud. It's primarily used for uploading large input data, output results, or log files that exceed the size limits for direct API calls.

RETURNS	DESCRIPTION
UploadURL	An object containing both the upload URL and an upload ID for reference. The upload URL is a pre-signed URL that allows temporary write access.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> # Get an upload URL and upload large input data
>>> upload_url = app.upload_url()
>>> large_input = {"locations": [...], "vehicles": [...]}
>>> app.upload_data(data=large_input, upload_url=upload_url)

Source code in nextmv/nextmv/cloud/application/__init__.py

def upload_url(self) -> UploadURL:
    """
    Get an upload URL to use for uploading a file.

    This method generates a pre-signed URL that can be used to upload large files
    to Nextmv Cloud. It's primarily used for uploading large input data, output
    results, or log files that exceed the size limits for direct API calls.

    Returns
    -------
    UploadURL
        An object containing both the upload URL and an upload ID for reference.
        The upload URL is a pre-signed URL that allows temporary write access.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> # Get an upload URL and upload large input data
    >>> upload_url = app.upload_url()
    >>> large_input = {"locations": [...], "vehicles": [...]}
    >>> app.upload_data(data=large_input, upload_url=upload_url)
    """

    response = self.client.request(
        method="POST",
        endpoint=f"{self.endpoint}/runs/uploadurl",
    )

    return UploadURL.from_dict(response.json())

ApplicationType

Bases: str, Enum

Enumeration of application types in Nextmv Cloud.

You can import the ApplicationType class directly from cloud:

from nextmv.cloud import ApplicationType

ATTRIBUTE	DESCRIPTION
CUSTOM	Custom application type, which is the most common. Represents a standard application that you can push code to. TYPE: str
SUBSCRIPTION	Subscription application type. You cannot push code to subscription applications, but only subscribe to them through the marketplace. TYPE: str
PIPELINE	Pipeline application type that refers to workflows. TYPE: str

Returned by:

• nextmv ☁️ Cloud Reference Application Module  application  Application  type

CUSTOM class-attribute instance-attribute

CUSTOM = 'custom'

Custom application type, which is the most common. Represents a standard application that you can push code to.

PIPELINE class-attribute instance-attribute

PIPELINE = 'pipeline'

Pipeline application type that refers to workflows.

SUBSCRIPTION class-attribute instance-attribute

SUBSCRIPTION = 'subscription'

Subscription application type. You cannot push code to subscription applications, but only subscribe to them through the marketplace.

list_applications

list_applications(client: Client) -> list[Application]

List all Nextmv Cloud applications.

You can import the list_applications function directly from cloud:

from nextmv.cloud import list_applications

PARAMETER	DESCRIPTION
client	The Nextmv Cloud client used to make API requests. TYPE: Client

RETURNS	DESCRIPTION
list[Application]	A list of Nextmv Cloud applications.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/__init__.py

def list_applications(client: Client) -> list[Application]:
    """
    List all Nextmv Cloud applications.

    You can import the `list_applications` function directly from `cloud`:

    ```python
    from nextmv.cloud import list_applications
    ```

    Parameters
    ----------
    client : Client
        The Nextmv Cloud client used to make API requests.

    Returns
    -------
    list[Application]
        A list of Nextmv Cloud applications.

    Raises
    -------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    response = client.request(
        method="GET",
        endpoint="v1/applications",
    )

    applications = []
    for app_data in response.json() or []:
        app = Application.from_dict({"client": client} | app_data)
        applications.append(app)

    return applications

Mixin Classes

These mixins extend the Application class with specific capabilities.

Acceptance Testing

_acceptance

Application mixin for managing acceptance tests.

ApplicationAcceptanceMixin

Mixin class providing acceptance test methods for Application.

acceptance_test

acceptance_test(acceptance_test_id: str) -> AcceptanceTest

Retrieve details of an acceptance test.

PARAMETER	DESCRIPTION
acceptance_test_id	ID of the acceptance test to retrieve. TYPE: str

RETURNS	DESCRIPTION
AcceptanceTest	The requested acceptance test details.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> test = app.acceptance_test("test-123")
>>> print(test.name)
'My Test'

Source code in nextmv/nextmv/cloud/application/_acceptance.py

def acceptance_test(self: "Application", acceptance_test_id: str) -> AcceptanceTest:
    """
    Retrieve details of an acceptance test.

    Parameters
    ----------
    acceptance_test_id : str
        ID of the acceptance test to retrieve.

    Returns
    -------
    AcceptanceTest
        The requested acceptance test details.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> test = app.acceptance_test("test-123")
    >>> print(test.name)
    'My Test'
    """
    response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/acceptance/{acceptance_test_id}",
    )

    return AcceptanceTest.from_dict(response.json())

acceptance_test_with_polling

acceptance_test_with_polling(
    acceptance_test_id: str,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
) -> AcceptanceTest

Retrieve details of an acceptance test using polling.

Retrieves the result of an acceptance test. This method polls for the result until the test finishes executing or the polling strategy is exhausted.

PARAMETER	DESCRIPTION
acceptance_test_id	ID of the acceptance test to retrieve. TYPE: str

RETURNS	DESCRIPTION
AcceptanceTest	The requested acceptance test details.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> test = app.acceptance_test_with_polling("test-123")
>>> print(test.name)
'My Test'

Source code in nextmv/nextmv/cloud/application/_acceptance.py

def acceptance_test_with_polling(
    self: "Application",
    acceptance_test_id: str,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
) -> AcceptanceTest:
    """
    Retrieve details of an acceptance test using polling.

    Retrieves the result of an acceptance test. This method polls for the
    result until the test finishes executing or the polling strategy is
    exhausted.

    Parameters
    ----------
    acceptance_test_id : str
        ID of the acceptance test to retrieve.

    Returns
    -------
    AcceptanceTest
        The requested acceptance test details.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> test = app.acceptance_test_with_polling("test-123")
    >>> print(test.name)
    'My Test'
    """

    def polling_func() -> tuple[Any, bool]:
        acceptance_test_result = self.acceptance_test(acceptance_test_id=acceptance_test_id)
        if acceptance_test_result.status in {
            ExperimentStatus.COMPLETED,
            ExperimentStatus.FAILED,
            ExperimentStatus.DRAFT,
            ExperimentStatus.CANCELED,
            ExperimentStatus.DELETE_FAILED,
        }:
            return acceptance_test_result, True

        return None, False

    acceptance_test = poll(polling_options=polling_options, polling_func=polling_func)

    return self.acceptance_test(acceptance_test_id=acceptance_test.id)

delete_acceptance_test

delete_acceptance_test(acceptance_test_id: str) -> None

Delete an acceptance test.

Deletes an acceptance test along with all the associated information such as the underlying batch experiment.

PARAMETER	DESCRIPTION
acceptance_test_id	ID of the acceptance test to delete. TYPE: str

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> app.delete_acceptance_test("test-123")

Source code in nextmv/nextmv/cloud/application/_acceptance.py

def delete_acceptance_test(self: "Application", acceptance_test_id: str) -> None:
    """
    Delete an acceptance test.

    Deletes an acceptance test along with all the associated information
    such as the underlying batch experiment.

    Parameters
    ----------
    acceptance_test_id : str
        ID of the acceptance test to delete.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> app.delete_acceptance_test("test-123")
    """

    _ = self.client.request(
        method="DELETE",
        endpoint=f"{self.experiments_endpoint}/acceptance/{acceptance_test_id}",
    )

list_acceptance_tests

list_acceptance_tests() -> list[AcceptanceTest]

List all acceptance tests.

RETURNS	DESCRIPTION
list[AcceptanceTest]	List of all acceptance tests associated with this application.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> tests = app.list_acceptance_tests()
>>> for test in tests:
...     print(test.name)
'Test 1'
'Test 2'

Source code in nextmv/nextmv/cloud/application/_acceptance.py

def list_acceptance_tests(self: "Application") -> list[AcceptanceTest]:
    """
    List all acceptance tests.

    Returns
    -------
    list[AcceptanceTest]
        List of all acceptance tests associated with this application.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> tests = app.list_acceptance_tests()
    >>> for test in tests:
    ...     print(test.name)
    'Test 1'
    'Test 2'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/acceptance",
    )

    return [AcceptanceTest.from_dict(acceptance_test) for acceptance_test in response.json()]

new_acceptance_test

new_acceptance_test(
    candidate_instance_id: str,
    baseline_instance_id: str,
    metrics: list[Metric | dict[str, Any]],
    id: str | None = None,
    name: str | None = None,
    input_set_id: str | None = None,
    description: str | None = None,
) -> AcceptanceTest

Create a new acceptance test.

The acceptance test is based on a batch experiment. If you already started a batch experiment, you don't need to provide the input_set_id parameter. In that case, the ID of the acceptance test and the batch experiment must be the same. If the batch experiment does not exist, you can provide the input_set_id parameter and a new batch experiment will be created for you.

PARAMETER	DESCRIPTION
candidate_instance_id	ID of the candidate instance. TYPE: str
baseline_instance_id	ID of the baseline instance. TYPE: str
id	ID of the acceptance test. Will be generated if not provided. TYPE: str | None DEFAULT: None
metrics	List of metrics to use for the acceptance test. TYPE: list[Union[Metric, dict[str, Any]]]
name	Name of the acceptance test. If not provided, the ID will be used as the name. TYPE: Optional[str] DEFAULT: None
input_set_id	ID of the input set to use for the underlying batch experiment, in case it hasn't been started. TYPE: Optional[str] DEFAULT: None
description	Description of the acceptance test. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
AcceptanceTest	The created acceptance test.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.
ValueError	If the batch experiment ID does not match the acceptance test ID.

Source code in nextmv/nextmv/cloud/application/_acceptance.py

def new_acceptance_test(  # noqa: C901
    self: "Application",
    candidate_instance_id: str,
    baseline_instance_id: str,
    metrics: list[Metric | dict[str, Any]],
    id: str | None = None,
    name: str | None = None,
    input_set_id: str | None = None,
    description: str | None = None,
) -> AcceptanceTest:
    """
    Create a new acceptance test.

    The acceptance test is based on a batch experiment. If you already
    started a batch experiment, you don't need to provide the input_set_id
    parameter. In that case, the ID of the acceptance test and the batch
    experiment must be the same. If the batch experiment does not exist,
    you can provide the input_set_id parameter and a new batch experiment
    will be created for you.

    Parameters
    ----------
    candidate_instance_id : str
        ID of the candidate instance.
    baseline_instance_id : str
        ID of the baseline instance.
    id : str | None, default=None
        ID of the acceptance test. Will be generated if not provided.
    metrics : list[Union[Metric, dict[str, Any]]]
        List of metrics to use for the acceptance test.
    name : Optional[str], default=None
        Name of the acceptance test. If not provided, the ID will be used as the name.
    input_set_id : Optional[str], default=None
        ID of the input set to use for the underlying batch experiment,
        in case it hasn't been started.
    description : Optional[str], default=None
        Description of the acceptance test.

    Returns
    -------
    AcceptanceTest
        The created acceptance test.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    ValueError
        If the batch experiment ID does not match the acceptance test ID.
    """

    # Generate ID if not provided
    if id is None or id == "":
        id = safe_id("acceptance")

    # Use ID as name if name not provided
    if name is None or name == "":
        name = id

    if input_set_id is None:
        try:
            batch_experiment = self.batch_experiment(batch_id=id)
            batch_experiment_id = batch_experiment.id
        except requests.HTTPError as e:
            if e.response.status_code != 404:
                raise e

            raise ValueError(
                f"batch experiment {id} does not exist, input_set_id must be defined to create a new one"
            ) from e
    else:
        # Get all input IDs from the input set.
        input_set = self.input_set(input_set_id=input_set_id)
        if not input_set.input_ids:
            raise ValueError(f"input set {input_set_id} does not contain any inputs")
        runs = []
        for input_id in input_set.input_ids:
            runs.append(
                BatchExperimentRun(
                    instance_id=candidate_instance_id,
                    input_set_id=input_set_id,
                    input_id=input_id,
                )
            )
            runs.append(
                BatchExperimentRun(
                    instance_id=baseline_instance_id,
                    input_set_id=input_set_id,
                    input_id=input_id,
                )
            )
        batch_experiment_id = self.new_batch_experiment(
            name=name,
            description=description,
            id=id,
            runs=runs,
        )

    if batch_experiment_id != id:
        raise ValueError(f"batch experiment_id ({batch_experiment_id}) does not match acceptance test id ({id})")

    payload_metrics = [{}] * len(metrics)
    for i, metric in enumerate(metrics):
        payload_metrics[i] = metric.to_dict() if isinstance(metric, Metric) else metric

    payload = {
        "candidate": {"instance_id": candidate_instance_id},
        "control": {"instance_id": baseline_instance_id},
        "metrics": payload_metrics,
        "experiment_id": batch_experiment_id,
        "name": name,
        "id": id,
    }
    if description is not None:
        payload["description"] = description

    response = self.client.request(
        method="POST",
        endpoint=f"{self.experiments_endpoint}/acceptance",
        payload=payload,
    )

    return AcceptanceTest.from_dict(response.json())

new_acceptance_test_with_result

new_acceptance_test_with_result(
    candidate_instance_id: str,
    baseline_instance_id: str,
    metrics: list[Metric | dict[str, Any]],
    id: str | None = None,
    name: str | None = None,
    input_set_id: str | None = None,
    description: str | None = None,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
) -> AcceptanceTest

Create a new acceptance test and poll for the result.

This is a convenience method that combines the new_acceptance_test with polling logic to check when the acceptance test is done.

PARAMETER	DESCRIPTION
candidate_instance_id	ID of the candidate instance. TYPE: str
baseline_instance_id	ID of the baseline instance. TYPE: str
id	ID of the acceptance test. Will be generated if not provided. TYPE: str | None DEFAULT: None
metrics	List of metrics to use for the acceptance test. TYPE: list[Union[Metric, dict[str, Any]]]
name	Name of the acceptance test. If not provided, the ID will be used as the name. TYPE: Optional[str] DEFAULT: None
input_set_id	ID of the input set to use for the underlying batch experiment, in case it hasn't been started. TYPE: Optional[str] DEFAULT: None
description	Description of the acceptance test. TYPE: Optional[str] DEFAULT: None
polling_options	Options to use when polling for the acceptance test result. TYPE: PollingOptions DEFAULT: _DEFAULT_POLLING_OPTIONS

RETURNS	DESCRIPTION
AcceptanceTest	The completed acceptance test with results.

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

Examples:

>>> test = app.new_acceptance_test_with_result(
...     candidate_instance_id="candidate-123",
...     baseline_instance_id="baseline-456",
...     id="test-789",
...     metrics=[Metric(name="objective", type="numeric")],
...     name="Performance Test",
...     input_set_id="input-set-123"
... )
>>> print(test.status)
'completed'

Source code in nextmv/nextmv/cloud/application/_acceptance.py

def new_acceptance_test_with_result(
    self: "Application",
    candidate_instance_id: str,
    baseline_instance_id: str,
    metrics: list[Metric | dict[str, Any]],
    id: str | None = None,
    name: str | None = None,
    input_set_id: str | None = None,
    description: str | None = None,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
) -> AcceptanceTest:
    """
    Create a new acceptance test and poll for the result.

    This is a convenience method that combines the new_acceptance_test with polling
    logic to check when the acceptance test is done.

    Parameters
    ----------
    candidate_instance_id : str
        ID of the candidate instance.
    baseline_instance_id : str
        ID of the baseline instance.
    id : str | None, default=None
        ID of the acceptance test. Will be generated if not provided.
    metrics : list[Union[Metric, dict[str, Any]]]
        List of metrics to use for the acceptance test.
    name : Optional[str], default=None
        Name of the acceptance test. If not provided, the ID will be used as the name.
    input_set_id : Optional[str], default=None
        ID of the input set to use for the underlying batch experiment,
        in case it hasn't been started.
    description : Optional[str], default=None
        Description of the acceptance test.
    polling_options : PollingOptions, default=_DEFAULT_POLLING_OPTIONS
        Options to use when polling for the acceptance test result.

    Returns
    -------
    AcceptanceTest
        The completed acceptance test with results.

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

    Examples
    --------
    >>> test = app.new_acceptance_test_with_result(
    ...     candidate_instance_id="candidate-123",
    ...     baseline_instance_id="baseline-456",
    ...     id="test-789",
    ...     metrics=[Metric(name="objective", type="numeric")],
    ...     name="Performance Test",
    ...     input_set_id="input-set-123"
    ... )
    >>> print(test.status)
    'completed'
    """

    acceptance_test = self.new_acceptance_test(
        candidate_instance_id=candidate_instance_id,
        baseline_instance_id=baseline_instance_id,
        id=id,
        metrics=metrics,
        name=name,
        input_set_id=input_set_id,
        description=description,
    )

    return self.acceptance_test_with_polling(
        acceptance_test_id=acceptance_test.id,
        polling_options=polling_options,
    )

update_acceptance_test

update_acceptance_test(
    acceptance_test_id: str,
    name: str | None = None,
    description: str | None = None,
) -> AcceptanceTest

Update an acceptance test.

PARAMETER	DESCRIPTION
acceptance_test_id	ID of the acceptance test to update. TYPE: str
name	Optional name of the acceptance test. TYPE: Optional[str] DEFAULT: None
description	Optional description of the acceptance test. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
AcceptanceTest	The updated acceptance test.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> test = app.update_acceptance_test(
...     acceptance_test_id="test-123",
...     name="Updated Test Name",
...     description="Updated description"
... )
>>> print(test.name)
'Updated Test Name'

Source code in nextmv/nextmv/cloud/application/_acceptance.py

def update_acceptance_test(
    self: "Application",
    acceptance_test_id: str,
    name: str | None = None,
    description: str | None = None,
) -> AcceptanceTest:
    """
    Update an acceptance test.

    Parameters
    ----------
    acceptance_test_id : str
        ID of the acceptance test to update.
    name : Optional[str], default=None
        Optional name of the acceptance test.
    description : Optional[str], default=None
        Optional description of the acceptance test.

    Returns
    -------
    AcceptanceTest
        The updated acceptance test.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> test = app.update_acceptance_test(
    ...     acceptance_test_id="test-123",
    ...     name="Updated Test Name",
    ...     description="Updated description"
    ... )
    >>> print(test.name)
    'Updated Test Name'
    """

    if (name is None or name == "") and (description is None or description == ""):
        raise ValueError("at least one of name or description must be provided for update")

    payload = {}

    if name is not None:
        payload["name"] = name
    if description is not None:
        payload["description"] = description

    response = self.client.request(
        method="PATCH",
        endpoint=f"{self.experiments_endpoint}/acceptance/{acceptance_test_id}",
        payload=payload,
    )

    return AcceptanceTest.from_dict(response.json())

Batch Scenarios

_batch_scenario

Application mixin for managing batch experiments.

ApplicationBatchMixin

Mixin class for managing batch experiments within an application.

batch_experiment

batch_experiment(batch_id: str) -> BatchExperiment

Get a batch experiment. This method also returns the runs of the batch experiment under the .runs attribute.

PARAMETER	DESCRIPTION
batch_id	ID of the batch experiment. TYPE: str

RETURNS	DESCRIPTION
BatchExperiment	The requested batch experiment details.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> batch_exp = app.batch_experiment("batch-123")
>>> print(batch_exp.name)
'My Batch Experiment'

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def batch_experiment(self: "Application", batch_id: str) -> BatchExperiment:
    """
    Get a batch experiment. This method also returns the runs of the batch
    experiment under the `.runs` attribute.

    Parameters
    ----------
    batch_id : str
        ID of the batch experiment.

    Returns
    -------
    BatchExperiment
        The requested batch experiment details.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> batch_exp = app.batch_experiment("batch-123")
    >>> print(batch_exp.name)
    'My Batch Experiment'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/batch/{batch_id}",
    )

    exp = BatchExperiment.from_dict(response.json())

    runs_response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/batch/{batch_id}/runs",
    )

    runs = [Run.from_dict(run) for run in runs_response.json().get("runs", [])]
    exp.runs = runs

    return exp

batch_experiment_metadata

batch_experiment_metadata(
    batch_id: str,
) -> BatchExperimentMetadata

Get metadata for a batch experiment.

PARAMETER	DESCRIPTION
batch_id	ID of the batch experiment. TYPE: str

RETURNS	DESCRIPTION
BatchExperimentMetadata	The requested batch experiment metadata.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> metadata = app.batch_experiment_metadata("batch-123")
>>> print(metadata.name)
'My Batch Experiment'

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def batch_experiment_metadata(self: "Application", batch_id: str) -> BatchExperimentMetadata:
    """
    Get metadata for a batch experiment.

    Parameters
    ----------
    batch_id : str
        ID of the batch experiment.

    Returns
    -------
    BatchExperimentMetadata
        The requested batch experiment metadata.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> metadata = app.batch_experiment_metadata("batch-123")
    >>> print(metadata.name)
    'My Batch Experiment'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/batch/{batch_id}/metadata",
    )

    return BatchExperimentMetadata.from_dict(response.json())

batch_experiment_with_polling

batch_experiment_with_polling(
    batch_id: str,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
) -> BatchExperiment

Get a batch experiment with polling.

Retrieves the result of an experiment. This method polls for the result until the experiment finishes executing or the polling strategy is exhausted.

PARAMETER	DESCRIPTION
batch_id	ID of the batch experiment. TYPE: str

RETURNS	DESCRIPTION
BatchExperiment	The requested batch experiment details.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> batch_exp = app.batch_experiment_with_polling("batch-123")
>>> print(batch_exp.name)
'My Batch Experiment'

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def batch_experiment_with_polling(
    self: "Application",
    batch_id: str,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
) -> BatchExperiment:
    """
    Get a batch experiment with polling.

    Retrieves the result of an experiment. This method polls for the result
    until the experiment finishes executing or the polling strategy is
    exhausted.

    Parameters
    ----------
    batch_id : str
        ID of the batch experiment.

    Returns
    -------
    BatchExperiment
        The requested batch experiment details.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> batch_exp = app.batch_experiment_with_polling("batch-123")
    >>> print(batch_exp.name)
    'My Batch Experiment'
    """

    def polling_func() -> tuple[Any, bool]:
        batch_metadata = self.batch_experiment_metadata(batch_id=batch_id)
        if batch_metadata.status in {
            ExperimentStatus.COMPLETED,
            ExperimentStatus.FAILED,
            ExperimentStatus.DRAFT,
            ExperimentStatus.CANCELED,
            ExperimentStatus.DELETE_FAILED,
        }:
            return batch_metadata, True

        return None, False

    batch_information = poll(polling_options=polling_options, polling_func=polling_func)

    return self.batch_experiment(batch_id=batch_information.id)

delete_batch_experiment

delete_batch_experiment(batch_id: str) -> None

Delete a batch experiment.

Deletes a batch experiment along with all the associated information, such as its runs.

PARAMETER	DESCRIPTION
batch_id	ID of the batch experiment to delete. TYPE: str

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> app.delete_batch_experiment("batch-123")

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def delete_batch_experiment(self: "Application", batch_id: str) -> None:
    """
    Delete a batch experiment.

    Deletes a batch experiment along with all the associated information,
    such as its runs.

    Parameters
    ----------
    batch_id : str
        ID of the batch experiment to delete.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> app.delete_batch_experiment("batch-123")
    """

    _ = self.client.request(
        method="DELETE",
        endpoint=f"{self.experiments_endpoint}/batch/{batch_id}",
    )

delete_scenario_test

delete_scenario_test(scenario_test_id: str) -> None

Delete a scenario test.

Deletes a scenario test. Scenario tests are based on the batch experiments API, so this function summons delete_batch_experiment.

PARAMETER	DESCRIPTION
scenario_test_id	ID of the scenario test to delete. TYPE: str

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> app.delete_scenario_test("scenario-123")

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def delete_scenario_test(self: "Application", scenario_test_id: str) -> None:
    """
    Delete a scenario test.

    Deletes a scenario test. Scenario tests are based on the batch
    experiments API, so this function summons `delete_batch_experiment`.

    Parameters
    ----------
    scenario_test_id : str
        ID of the scenario test to delete.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> app.delete_scenario_test("scenario-123")
    """

    self.delete_batch_experiment(batch_id=scenario_test_id)

list_batch_experiments

list_batch_experiments() -> list[BatchExperimentMetadata]

List all batch experiments.

RETURNS	DESCRIPTION
list[BatchExperimentMetadata]	List of batch experiments.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def list_batch_experiments(self: "Application") -> list[BatchExperimentMetadata]:
    """
    List all batch experiments.

    Returns
    -------
    list[BatchExperimentMetadata]
        List of batch experiments.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/batch",
        query_params={"type": "batch"},
    )

    return [BatchExperimentMetadata.from_dict(batch_experiment) for batch_experiment in response.json()]

list_scenario_tests

list_scenario_tests() -> list[BatchExperimentMetadata]

List all batch scenario tests. Scenario tests are based on the batch experiments API, so this function returns the same information as list_batch_experiments, albeit using a different query parameter.

RETURNS	DESCRIPTION
list[BatchExperimentMetadata]	List of scenario tests.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def list_scenario_tests(self: "Application") -> list[BatchExperimentMetadata]:
    """
    List all batch scenario tests. Scenario tests are based on the batch
    experiments API, so this function returns the same information as
    `list_batch_experiments`, albeit using a different query parameter.

    Returns
    -------
    list[BatchExperimentMetadata]
        List of scenario tests.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/batch",
        query_params={"type": "scenario"},
    )

    return [BatchExperimentMetadata.from_dict(batch_experiment) for batch_experiment in response.json()]

new_batch_experiment

new_batch_experiment(
    name: str | None = None,
    input_set_id: str | None = None,
    description: str | None = None,
    id: str | None = None,
    option_sets: dict[str, dict[str, str]] | None = None,
    runs: list[BatchExperimentRun | dict[str, Any]]
    | None = None,
    type: str | None = "batch",
) -> str

Create a new batch experiment.

PARAMETER	DESCRIPTION
name	Name of the batch experiment. If not provided, the ID will be used as the name. TYPE: str | None DEFAULT: None
input_set_id	ID of the input set to use for the batch experiment. TYPE: str | None DEFAULT: None
description	Optional description of the batch experiment. TYPE: str | None DEFAULT: None
id	ID of the batch experiment. Will be generated if not provided. TYPE: str | None DEFAULT: None
option_sets	Option sets to use for the batch experiment. This is a dictionary where the keys are option set IDs and the values are dictionaries with the actual options. TYPE: dict[str, dict[str, str]] | None DEFAULT: None
runs	List of runs to use for the batch experiment. TYPE: list[BatchExperimentRun | dict[str, Any]] | None DEFAULT: None
type	Type of the batch experiment. This is used to determine the experiment type. The default value is "batch". If you want to create a scenario test, set this to "scenario". TYPE: str | None DEFAULT: 'batch'

RETURNS	DESCRIPTION
str	ID of the batch experiment.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def new_batch_experiment(  # noqa: C901
    self: "Application",
    name: str | None = None,
    input_set_id: str | None = None,
    description: str | None = None,
    id: str | None = None,
    option_sets: dict[str, dict[str, str]] | None = None,
    runs: list[BatchExperimentRun | dict[str, Any]] | None = None,
    type: str | None = "batch",
) -> str:
    """
    Create a new batch experiment.

    Parameters
    ----------
    name: Optional[str]
        Name of the batch experiment. If not provided, the ID will be used as the name.
    input_set_id: str | None
        ID of the input set to use for the batch experiment.
    description: Optional[str]
        Optional description of the batch experiment.
    id: Optional[str]
        ID of the batch experiment. Will be generated if not provided.
    option_sets: Optional[dict[str, dict[str, str]]]
        Option sets to use for the batch experiment. This is a dictionary
        where the keys are option set IDs and the values are dictionaries
        with the actual options.
    runs: Optional[list[BatchExperimentRun]]
        List of runs to use for the batch experiment.
    type: Optional[str]
        Type of the batch experiment. This is used to determine the
        experiment type. The default value is "batch". If you want to
        create a scenario test, set this to "scenario".

    Returns
    -------
    str
        ID of the batch experiment.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    # Generate ID if not provided
    if id is None or id == "":
        id = safe_id("batch")

    # Use ID as name if name not provided
    if name is None or name == "":
        name = id

    payload = {
        "id": id,
        "name": name,
    }
    if input_set_id is not None:
        payload["input_set_id"] = input_set_id
    if description is not None:
        payload["description"] = description
    if option_sets is not None:
        payload["option_sets"] = option_sets
    if runs is not None:
        payload_runs = [{}] * len(runs)
        for i, run in enumerate(runs):
            payload_runs[i] = run.to_dict() if isinstance(run, BatchExperimentRun) else run
        payload["runs"] = payload_runs
    if type is not None:
        payload["type"] = type

    response = self.client.request(
        method="POST",
        endpoint=f"{self.experiments_endpoint}/batch",
        payload=payload,
    )

    return response.json()["id"]

new_batch_experiment_with_result

new_batch_experiment_with_result(
    name: str | None = None,
    input_set_id: str | None = None,
    description: str | None = None,
    id: str | None = None,
    option_sets: dict[str, dict[str, str]] | None = None,
    runs: list[BatchExperimentRun | dict[str, Any]]
    | None = None,
    type: str | None = "batch",
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
) -> BatchExperiment

Convenience method to create a new batch experiment and poll for the result.

This method combines the new_batch_experiment and batch_experiment_with_polling methods, applying polling logic to check when the experiment succeeded.

PARAMETER	DESCRIPTION
name	Name of the batch experiment. If not provided, the ID will be used as the name. TYPE: str | None DEFAULT: None
input_set_id	ID of the input set to use for the batch experiment. TYPE: str | None DEFAULT: None
description	Optional description of the batch experiment. TYPE: str | None DEFAULT: None
id	ID of the batch experiment. Will be generated if not provided. TYPE: str | None DEFAULT: None
option_sets	Option sets to use for the batch experiment. This is a dictionary where the keys are option set IDs and the values are dictionaries with the actual options. TYPE: dict[str, dict[str, str]] | None DEFAULT: None
runs	List of runs to use for the batch experiment. TYPE: list[BatchExperimentRun | dict[str, Any]] | None DEFAULT: None
type	Type of the batch experiment. This is used to determine the experiment type. The default value is "batch". If you want to create a scenario test, set this to "scenario". TYPE: str | None DEFAULT: 'batch'
polling_options	Options to use when polling for the batch experiment result. TYPE: PollingOptions DEFAULT: DEFAULT_POLLING_OPTIONS

RETURNS	DESCRIPTION
BatchExperiment	The completed batch experiment with results.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def new_batch_experiment_with_result(
    self: "Application",
    name: str | None = None,
    input_set_id: str | None = None,
    description: str | None = None,
    id: str | None = None,
    option_sets: dict[str, dict[str, str]] | None = None,
    runs: list[BatchExperimentRun | dict[str, Any]] | None = None,
    type: str | None = "batch",
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
) -> BatchExperiment:
    """
    Convenience method to create a new batch experiment and poll for the
    result.

    This method combines the `new_batch_experiment` and
    `batch_experiment_with_polling` methods, applying polling logic to
    check when the experiment succeeded.

    Parameters
    ----------
    name: Optional[str]
        Name of the batch experiment. If not provided, the ID will be used as the name.
    input_set_id: str
        ID of the input set to use for the batch experiment.
    description: Optional[str]
        Optional description of the batch experiment.
    id: Optional[str]
        ID of the batch experiment. Will be generated if not provided.
    option_sets: Optional[dict[str, dict[str, str]]]
        Option sets to use for the batch experiment. This is a dictionary
        where the keys are option set IDs and the values are dictionaries
        with the actual options.
    runs: Optional[list[BatchExperimentRun]]
        List of runs to use for the batch experiment.
    type: Optional[str]
        Type of the batch experiment. This is used to determine the
        experiment type. The default value is "batch". If you want to
        create a scenario test, set this to "scenario".
    polling_options : PollingOptions, default=DEFAULT_POLLING_OPTIONS
        Options to use when polling for the batch experiment result.

    Returns
    -------
    BatchExperiment
        The completed batch experiment with results.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    batch_id = self.new_batch_experiment(
        name=name,
        input_set_id=input_set_id,
        description=description,
        id=id,
        option_sets=option_sets,
        runs=runs,
        type=type,
    )

    return self.batch_experiment_with_polling(batch_id=batch_id, polling_options=polling_options)

new_scenario_test

new_scenario_test(
    scenarios: list[Scenario],
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    repetitions: int | None = 0,
) -> str

Create a new scenario test. The test is based on scenarios and you may specify repetitions to run the test multiple times. 0 repetitions means that the tests will be executed once. 1 repetition means that the test will be repeated once, i.e.: it will be executed twice. 2 repetitions equals 3 executions, so on, and so forth.

For each scenario, consider the scenario_input and configuration. The scenario_input.scenario_input_type allows you to specify the data that will be used for that scenario.

• ScenarioInputType.INPUT_SET: the data should be taken from an existing input set.
• ScenarioInputType.INPUT: the data should be taken from a list of existing inputs. When using this type, an input set will be created from this set of managed inputs.
• ScenarioInputType.New: a new set of data will be uploaded as a set of managed inputs. A new input set will be created from this set of managed inputs.

On the other hand, the configuration allows you to specify multiple option variations for the scenario. Please see the ScenarioConfiguration class for more information.

The scenario tests uses the batch experiments API under the hood.

PARAMETER	DESCRIPTION
scenarios	List of scenarios to use for the scenario test. At least one scenario should be provided. TYPE: list[Scenario]
id	ID of the scenario test. Will be generated if not provided. TYPE: str | None DEFAULT: None
name	Name of the scenario test. If not provided, the ID will be used as the name. TYPE: str | None DEFAULT: None
description	Optional description of the scenario test. TYPE: str | None DEFAULT: None
repetitions	Number of repetitions to use for the scenario test. 0 repetitions means that the tests will be executed once. 1 repetition means that the test will be repeated once, i.e.: it will be executed twice. 2 repetitions equals 3 executions, so on, and so forth. TYPE: int | None DEFAULT: 0

RETURNS	DESCRIPTION
str	ID of the scenario test.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.
ValueError	If no scenarios are provided.

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def new_scenario_test(
    self: "Application",
    scenarios: list[Scenario],
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    repetitions: int | None = 0,
) -> str:
    """
    Create a new scenario test. The test is based on `scenarios` and you
    may specify `repetitions` to run the test multiple times. 0 repetitions
    means that the tests will be executed once. 1 repetition means that the
    test will be repeated once, i.e.: it will be executed twice. 2
    repetitions equals 3 executions, so on, and so forth.

    For each scenario, consider the `scenario_input` and `configuration`.
    The `scenario_input.scenario_input_type` allows you to specify the data
    that will be used for that scenario.

    - `ScenarioInputType.INPUT_SET`: the data should be taken from an
      existing input set.
    - `ScenarioInputType.INPUT`: the data should be taken from a list of
      existing inputs. When using this type, an input set will be created
      from this set of managed inputs.
    - `ScenarioInputType.New`: a new set of data will be uploaded as a set
      of managed inputs. A new input set will be created from this set of
      managed inputs.

    On the other hand, the `configuration` allows you to specify multiple
    option variations for the scenario. Please see the
    `ScenarioConfiguration` class for more information.

    The scenario tests uses the batch experiments API under the hood.

    Parameters
    ----------
    scenarios: list[Scenario]
        List of scenarios to use for the scenario test. At least one
        scenario should be provided.
    id: Optional[str]
        ID of the scenario test. Will be generated if not provided.
    name: Optional[str]
        Name of the scenario test. If not provided, the ID will be used as the name.
    description: Optional[str]
        Optional description of the scenario test.
    repetitions: Optional[int]
        Number of repetitions to use for the scenario test. 0
        repetitions means that the tests will be executed once. 1
        repetition means that the test will be repeated once, i.e.: it
        will be executed twice. 2 repetitions equals 3 executions, so on,
        and so forth.

    Returns
    -------
    str
        ID of the scenario test.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    ValueError
        If no scenarios are provided.
    """

    if len(scenarios) < 1:
        raise ValueError("At least one scenario must be provided")

    # Generate ID if not provided
    if id is None or id == "":
        id = safe_id("scenario")

    # Use ID as name if name not provided
    if name is None or name == "":
        name = id

    scenarios_by_id = _scenarios_by_id(scenarios)

    # Save all the information needed by scenario.
    input_sets = {}
    instances = {}
    for scenario_id, scenario in scenarios_by_id.items():
        instance = self.instance(instance_id=scenario.instance_id)

        # Each scenario is associated to an input set, so we must either
        # get it or create it.
        input_set = self.__input_set_for_scenario(scenario, scenario_id)

        instances[scenario_id] = instance
        input_sets[scenario_id] = input_set

    # Calculate the combinations of all the option sets across scenarios.
    opt_sets_by_scenario = _option_sets(scenarios)

    # The scenario tests results in multiple individual runs.
    runs = []
    run_counter = 0
    opt_sets = {}
    for scenario_id, scenario_opt_sets in opt_sets_by_scenario.items():
        opt_sets = {**opt_sets, **scenario_opt_sets}
        input_set = input_sets[scenario_id]
        scenario = scenarios_by_id[scenario_id]

        for set_key in scenario_opt_sets.keys():
            inputs = input_set.input_ids if len(input_set.input_ids) > 0 else input_set.inputs
            for input in inputs:
                input_id = input.id if isinstance(input, ManagedInput) else input
                for repetition in range(repetitions + 1):
                    run_counter += 1
                    run = BatchExperimentRun(
                        input_id=input_id,
                        input_set_id=input_set.id,
                        instance_id=scenario.instance_id,
                        option_set=set_key,
                        scenario_id=scenario_id,
                        repetition=repetition,
                        run_number=f"{run_counter}",
                    )
                    runs.append(run)

    return self.new_batch_experiment(
        id=id,
        name=name,
        description=description,
        type="scenario",
        option_sets=opt_sets,
        runs=runs,
    )

new_scenario_test_with_result

new_scenario_test_with_result(
    scenarios: list[Scenario],
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    repetitions: int | None = 0,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
) -> BatchExperiment

Convenience method to create a new scenario test and poll for the result.

This method combines the new_scenario_test and scenario_test_with_polling methods, applying polling logic to check when the test succeeded.

The scenario tests uses the batch experiments API under the hood.

PARAMETER	DESCRIPTION
scenarios	List of scenarios to use for the scenario test. At least one scenario should be provided. TYPE: list[Scenario]
id	ID of the scenario test. Will be generated if not provided. TYPE: str | None DEFAULT: None
name	Name of the scenario test. If not provided, the ID will be used as the name. TYPE: str | None DEFAULT: None
description	Optional description of the scenario test. TYPE: str | None DEFAULT: None
repetitions	Number of repetitions to use for the scenario test. 0 repetitions means that the tests will be executed once. 1 repetition means that the test will be repeated once, i.e.: it will be executed twice. 2 repetitions equals 3 executions, so on, and so forth. TYPE: int | None DEFAULT: 0
polling_options	Options to use when polling for the scenario test result. TYPE: PollingOptions DEFAULT: DEFAULT_POLLING_OPTIONS

RETURNS	DESCRIPTION
BatchExperiment	The completed scenario test as a BatchExperiment.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.
ValueError	If no scenarios are provided.

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def new_scenario_test_with_result(
    self: "Application",
    scenarios: list[Scenario],
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    repetitions: int | None = 0,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
) -> BatchExperiment:
    """
    Convenience method to create a new scenario test and poll for the
    result.

    This method combines the `new_scenario_test` and
    `scenario_test_with_polling` methods, applying polling logic to
    check when the test succeeded.

    The scenario tests uses the batch experiments API under the hood.

    Parameters
    ----------
    scenarios: list[Scenario]
        List of scenarios to use for the scenario test. At least one
        scenario should be provided.
    id: Optional[str]
        ID of the scenario test. Will be generated if not provided.
    name: Optional[str]
        Name of the scenario test. If not provided, the ID will be used as the name.
    description: Optional[str]
        Optional description of the scenario test.
    repetitions: Optional[int]
        Number of repetitions to use for the scenario test. 0
        repetitions means that the tests will be executed once. 1
        repetition means that the test will be repeated once, i.e.: it
        will be executed twice. 2 repetitions equals 3 executions, so on,
        and so forth.
    polling_options : PollingOptions, default=DEFAULT_POLLING_OPTIONS
        Options to use when polling for the scenario test result.

    Returns
    -------
    BatchExperiment
        The completed scenario test as a BatchExperiment.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    ValueError
        If no scenarios are provided.
    """

    test_id = self.new_scenario_test(
        scenarios=scenarios,
        id=id,
        name=name,
        description=description,
        repetitions=repetitions,
    )

    return self.scenario_test_with_polling(
        scenario_test_id=test_id,
        polling_options=polling_options,
    )

scenario_test

scenario_test(scenario_test_id: str) -> BatchExperiment

Get a scenario test.

Retrieves a scenario test by ID. Scenario tests are based on batch experiments, so this function returns the corresponding batch experiment associated with the scenario test.

PARAMETER	DESCRIPTION
scenario_test_id	ID of the scenario test to retrieve. TYPE: str

RETURNS	DESCRIPTION
BatchExperiment	The scenario test details as a batch experiment.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> test = app.scenario_test("scenario-123")
>>> print(test.name)
'My Scenario Test'
>>> print(test.type)
'scenario'

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def scenario_test(self: "Application", scenario_test_id: str) -> BatchExperiment:
    """
    Get a scenario test.

    Retrieves a scenario test by ID. Scenario tests are based on batch
    experiments, so this function returns the corresponding batch
    experiment associated with the scenario test.

    Parameters
    ----------
    scenario_test_id : str
        ID of the scenario test to retrieve.

    Returns
    -------
    BatchExperiment
        The scenario test details as a batch experiment.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> test = app.scenario_test("scenario-123")
    >>> print(test.name)
    'My Scenario Test'
    >>> print(test.type)
    'scenario'
    """

    return self.batch_experiment(batch_id=scenario_test_id)

scenario_test_metadata

scenario_test_metadata(
    scenario_test_id: str,
) -> BatchExperimentMetadata

Get the metadata for a scenario test, given its ID.

Scenario tests are based on batch experiments, so this function returns the corresponding batch experiment metadata associated with the scenario test.

PARAMETER	DESCRIPTION
scenario_test_id	ID of the scenario test to retrieve. TYPE: str

RETURNS	DESCRIPTION
BatchExperimentMetadata	The scenario test metadata as a batch experiment.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> metadata = app.scenario_test_metadata("scenario-123")
>>> print(metadata.name)
'My Scenario Test'
>>> print(metadata.type)
'scenario'

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def scenario_test_metadata(self: "Application", scenario_test_id: str) -> BatchExperimentMetadata:
    """
    Get the metadata for a scenario test, given its ID.

    Scenario tests are based on batch experiments, so this function returns
    the corresponding batch experiment metadata associated with the
    scenario test.

    Parameters
    ----------
    scenario_test_id : str
        ID of the scenario test to retrieve.

    Returns
    -------
    BatchExperimentMetadata
        The scenario test metadata as a batch experiment.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> metadata = app.scenario_test_metadata("scenario-123")
    >>> print(metadata.name)
    'My Scenario Test'
    >>> print(metadata.type)
    'scenario'
    """

    return self.batch_experiment_metadata(batch_id=scenario_test_id)

scenario_test_with_polling

scenario_test_with_polling(
    scenario_test_id: str,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
) -> BatchExperiment

Get a scenario test with polling.

Retrieves the result of a scenario test. This method polls for the result until the test finishes executing or the polling strategy is exhausted.

The scenario tests uses the batch experiments API under the hood.

PARAMETER	DESCRIPTION
scenario_test_id	ID of the scenario test to retrieve. TYPE: str
polling_options	Options to use when polling for the scenario test result. TYPE: PollingOptions DEFAULT: DEFAULT_POLLING_OPTIONS

RETURNS	DESCRIPTION
BatchExperiment	The scenario test details as a batch experiment.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> test = app.scenario_test_with_polling("scenario-123")
>>> print(test.name)
'My Scenario Test'
>>> print(test.type)
'scenario'

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def scenario_test_with_polling(
    self: "Application",
    scenario_test_id: str,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
) -> BatchExperiment:
    """
    Get a scenario test with polling.

    Retrieves the result of a scenario test. This method polls for the
    result until the test finishes executing or the polling strategy is
    exhausted.

    The scenario tests uses the batch experiments API under the hood.

    Parameters
    ----------
    scenario_test_id : str
        ID of the scenario test to retrieve.
    polling_options : PollingOptions, default=DEFAULT_POLLING_OPTIONS
        Options to use when polling for the scenario test result.

    Returns
    -------
    BatchExperiment
        The scenario test details as a batch experiment.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> test = app.scenario_test_with_polling("scenario-123")
    >>> print(test.name)
    'My Scenario Test'
    >>> print(test.type)
    'scenario'
    """

    return self.batch_experiment_with_polling(batch_id=scenario_test_id, polling_options=polling_options)

update_batch_experiment

update_batch_experiment(
    batch_experiment_id: str,
    name: str | None = None,
    description: str | None = None,
) -> BatchExperimentInformation

Update a batch experiment.

PARAMETER	DESCRIPTION
batch_experiment_id	ID of the batch experiment to update. TYPE: str
name	Optional name of the batch experiment. TYPE: Optional[str] DEFAULT: None
description	Optional description of the batch experiment. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
BatchExperimentInformation	The information with the updated batch experiment.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def update_batch_experiment(
    self: "Application",
    batch_experiment_id: str,
    name: str | None = None,
    description: str | None = None,
) -> BatchExperimentInformation:
    """
    Update a batch experiment.

    Parameters
    ----------
    batch_experiment_id : str
        ID of the batch experiment to update.
    name : Optional[str], default=None
        Optional name of the batch experiment.
    description : Optional[str], default=None
        Optional description of the batch experiment.

    Returns
    -------
    BatchExperimentInformation
        The information with the updated batch experiment.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    payload = {}

    if name is not None:
        payload["name"] = name
    if description is not None:
        payload["description"] = description

    response = self.client.request(
        method="PATCH",
        endpoint=f"{self.experiments_endpoint}/batch/{batch_experiment_id}",
        payload=payload,
    )

    return BatchExperimentInformation.from_dict(response.json())

update_scenario_test

update_scenario_test(
    scenario_test_id: str,
    name: str | None = None,
    description: str | None = None,
) -> BatchExperimentInformation

Update a scenario test.

Updates a scenario test with new name and description. Scenario tests use the batch experiments API, so this method calls the update_batch_experiment method, and thus the return type is the same.

PARAMETER	DESCRIPTION
scenario_test_id	ID of the scenario test to update. TYPE: str
name	Optional new name for the scenario test. TYPE: Optional[str] DEFAULT: None
description	Optional new description for the scenario test. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
BatchExperimentInformation	The information about the updated scenario test.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> info = app.update_scenario_test(
...     scenario_test_id="scenario-123",
...     name="Updated Test Name",
...     description="Updated description for this test"
... )
>>> print(info.name)
'Updated Test Name'

Source code in nextmv/nextmv/cloud/application/_batch_scenario.py

def update_scenario_test(
    self: "Application",
    scenario_test_id: str,
    name: str | None = None,
    description: str | None = None,
) -> BatchExperimentInformation:
    """
    Update a scenario test.

    Updates a scenario test with new name and description. Scenario tests
    use the batch experiments API, so this method calls the
    `update_batch_experiment` method, and thus the return type is the same.

    Parameters
    ----------
    scenario_test_id : str
        ID of the scenario test to update.
    name : Optional[str], default=None
        Optional new name for the scenario test.
    description : Optional[str], default=None
        Optional new description for the scenario test.

    Returns
    -------
    BatchExperimentInformation
        The information about the updated scenario test.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> info = app.update_scenario_test(
    ...     scenario_test_id="scenario-123",
    ...     name="Updated Test Name",
    ...     description="Updated description for this test"
    ... )
    >>> print(info.name)
    'Updated Test Name'
    """

    return self.update_batch_experiment(
        batch_experiment_id=scenario_test_id,
        name=name,
        description=description,
    )

Ensemble Management

_ensemble

Application mixin for managing app ensembles.

ApplicationEnsembleMixin

Mixin class for managing app ensembles within an application.

delete_ensemble_definition

delete_ensemble_definition(
    ensemble_definition_id: str,
) -> None

Delete an ensemble definition.

PARAMETER	DESCRIPTION
ensemble_definition_id	ID of the ensemble definition to delete. TYPE: str

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> app.delete_ensemble_definition("development-ensemble-definition")

Source code in nextmv/nextmv/cloud/application/_ensemble.py

def delete_ensemble_definition(self: "Application", ensemble_definition_id: str) -> None:
    """
    Delete an ensemble definition.

    Parameters
    ----------
    ensemble_definition_id : str
        ID of the ensemble definition to delete.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> app.delete_ensemble_definition("development-ensemble-definition")
    """

    _ = self.client.request(
        method="DELETE",
        endpoint=f"{self.ensembles_endpoint}/{ensemble_definition_id}",
    )

ensemble_definition

ensemble_definition(
    ensemble_definition_id: str,
) -> EnsembleDefinition

Get an ensemble definition.

PARAMETER	DESCRIPTION
ensemble_definition_id	ID of the ensemble definition to retrieve. TYPE: str

RETURNS	DESCRIPTION
EnsembleDefintion	The requested ensemble definition details.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> ensemble_definition = app.ensemble_definition("instance-123")
>>> print(ensemble_definition.name)
'Production Ensemble Definition'

Source code in nextmv/nextmv/cloud/application/_ensemble.py

def ensemble_definition(self: "Application", ensemble_definition_id: str) -> EnsembleDefinition:
    """
    Get an ensemble definition.

    Parameters
    ----------
    ensemble_definition_id : str
        ID of the ensemble definition to retrieve.

    Returns
    -------
    EnsembleDefintion
        The requested ensemble definition details.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> ensemble_definition = app.ensemble_definition("instance-123")
    >>> print(ensemble_definition.name)
    'Production Ensemble Definition'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.ensembles_endpoint}/{ensemble_definition_id}",
    )

    return EnsembleDefinition.from_dict(response.json())

list_ensemble_definitions

list_ensemble_definitions() -> list[EnsembleDefinition]

List all ensemble_definitions.

RETURNS	DESCRIPTION
list[EnsembleDefinition]	List of all ensemble definitions associated with this application.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> ensemble_definitions = app.list_ensemble_definitions()
>>> for ensemble_definition in ensemble_definitions:
...     print(ensemble_definition.name)
'Development Ensemble Definition'
'Production Ensemble Definition'

Source code in nextmv/nextmv/cloud/application/_ensemble.py

def list_ensemble_definitions(self: "Application") -> list[EnsembleDefinition]:
    """
    List all ensemble_definitions.

    Returns
    -------
    list[EnsembleDefinition]
        List of all ensemble definitions associated with this application.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> ensemble_definitions = app.list_ensemble_definitions()
    >>> for ensemble_definition in ensemble_definitions:
    ...     print(ensemble_definition.name)
    'Development Ensemble Definition'
    'Production Ensemble Definition'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.ensembles_endpoint}",
    )

    return [EnsembleDefinition.from_dict(ensemble_definition) for ensemble_definition in response.json()["items"]]

new_ensemble_definition

new_ensemble_definition(
    run_groups: list[RunGroup],
    rules: list[EvaluationRule],
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
) -> EnsembleDefinition

Create a new ensemble definition.

PARAMETER	DESCRIPTION
run_groups	Information to facilitate the execution of child runs. TYPE: list[RunGroup]
rules	Information to facilitate the selection of a result for the ensemble run from child runs. TYPE: list[EvaluationRule]
id	ID of the ensemble definition. If not provided, a unique ID will be generated with the prefix 'ensemble-'. TYPE: str | None DEFAULT: None
name	Name of the ensemble definition. If not provided, the ID will be used. TYPE: str | None DEFAULT: None
description	Description of the ensemble definition. If not provided, the name will be used. TYPE: str | None DEFAULT: None

Source code in nextmv/nextmv/cloud/application/_ensemble.py

def new_ensemble_definition(
    self: "Application",
    run_groups: list[RunGroup],
    rules: list[EvaluationRule],
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
) -> EnsembleDefinition:
    """
    Create a new ensemble definition.

    Parameters
    ----------
    run_groups: list[RunGroup]
        Information to facilitate the execution of child runs.
    rules: list[EvaluationRule]
        Information to facilitate the selection of
        a result for the ensemble run from child runs.
    id: str | None, default=None
        ID of the ensemble definition. If not provided, a unique ID will be
        generated with the prefix 'ensemble-'.
    name: Optional[str]
        Name of the ensemble definition. If not provided, the ID will be used.
    description: Optional[str]
        Description of the ensemble definition. If not provided, the name will be used.
    """

    if len(run_groups) == 0:
        raise ValueError("at least one run group must be defined to create an ensemble definition")

    if len(rules) == 0:
        raise ValueError("at least one evaluation rule must be defined to create an ensemble definition")

    if id is None or id == "":
        id = safe_id(prefix="ensemble")
    if name is None or name == "":
        name = id
    if description is None or description == "":
        description = name

    payload = {
        "id": id,
        "run_groups": [run_group.to_dict() for run_group in run_groups],
        "rules": [rule.to_dict() for rule in rules],
        "name": name,
        "description": description,
    }

    response = self.client.request(
        method="POST",
        endpoint=f"{self.ensembles_endpoint}",
        payload=payload,
    )

    return EnsembleDefinition.from_dict(response.json())

new_ensemble_defintion

new_ensemble_defintion(
    id: str,
    run_groups: list[RunGroup],
    rules: list[EvaluationRule],
    name: str | None = None,
    description: str | None = None,
) -> EnsembleDefinition

Warning

new_ensemble_defintion is deprecated, use new_ensemble_definition instead.

Create a new ensemble definition.

PARAMETER	DESCRIPTION
id	ID of the ensemble defintion. TYPE: str
run_groups	Information to facilitate the execution of child runs. TYPE: list[RunGroup]
rules	Information to facilitate the selection of a result for the ensemble run from child runs. TYPE: list[EvaluationRule]
name	Name of the ensemble definition. TYPE: str | None DEFAULT: None
description	Description of the ensemble definition. TYPE: str | None DEFAULT: None

Source code in nextmv/nextmv/cloud/application/_ensemble.py

def new_ensemble_defintion(
    self: "Application",
    id: str,
    run_groups: list[RunGroup],
    rules: list[EvaluationRule],
    name: str | None = None,
    description: str | None = None,
) -> EnsembleDefinition:
    """
    !!! warning
        `new_ensemble_defintion` is deprecated, use `new_ensemble_definition` instead.

    Create a new ensemble definition.

    Parameters
    ----------
    id: str
        ID of the ensemble defintion.
    run_groups: list[RunGroup]
        Information to facilitate the execution of child runs.
    rules: list[EvaluationRule]
        Information to facilitate the selection of
        a result for the ensemble run from child runs.
    name: Optional[str]
        Name of the ensemble definition.
    description: Optional[str]
        Description of the ensemble definition.
    """

    deprecated(
        name="new_ensemble_defintion",
        reason="`Application.new_ensemble_defintion` is deprecated, use `new_ensemble_definition` instead",
    )

    return self.new_ensemble_definition(
        run_groups=run_groups,
        rules=rules,
        id=id,
        name=name,
        description=description,
    )

update_ensemble_definition

update_ensemble_definition(
    id: str,
    name: str | None = None,
    description: str | None = None,
) -> EnsembleDefinition

Update an ensemble definition.

PARAMETER	DESCRIPTION
id	ID of the ensemble definition to update. TYPE: str
name	Optional name of the ensemble definition. TYPE: Optional[str] DEFAULT: None
description	Optional description of the ensemble definition. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
EnsembleDefinition	The updated ensemble definition.

RAISES	DESCRIPTION
ValueError	If neither name nor description is updated
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_ensemble.py

def update_ensemble_definition(
    self: "Application",
    id: str,
    name: str | None = None,
    description: str | None = None,
) -> EnsembleDefinition:
    """
    Update an ensemble definition.

    Parameters
    ----------
    id : str
        ID of the ensemble definition to update.
    name : Optional[str], default=None
        Optional name of the ensemble definition.
    description : Optional[str], default=None
        Optional description of the ensemble definition.

    Returns
    -------
    EnsembleDefinition
        The updated ensemble definition.

    Raises
    ------
    ValueError
        If neither name nor description is updated
    requests.HTTPError
        If the response status code is not 2xx.
    """

    payload = {}

    if name is None and description is None:
        raise ValueError("Must define at least one value among name and description to modify")
    if name is not None:
        payload["name"] = name
    if description is not None:
        payload["description"] = description

    response = self.client.request(
        method="PATCH",
        endpoint=f"{self.ensembles_endpoint}/{id}",
        payload=payload,
    )

    return EnsembleDefinition.from_dict(response.json())

Input Sets

_input_set

Application mixin for managing app input sets.

ApplicationInputSetMixin

Mixin class for managing app input sets within an application.

delete_input_set

delete_input_set(input_set_id: str) -> None

Delete an input set.

Deletes an input set along with all the associated information.

PARAMETER	DESCRIPTION
input_set_id	ID of the input set to delete. TYPE: str

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> app.delete_input_set("input-set-123")

Source code in nextmv/nextmv/cloud/application/_input_set.py

def delete_input_set(self: "Application", input_set_id: str) -> None:
    """
    Delete an input set.

    Deletes an input set along with all the associated information.

    Parameters
    ----------
    input_set_id : str
        ID of the input set to delete.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> app.delete_input_set("input-set-123")
    """

    _ = self.client.request(
        method="DELETE",
        endpoint=f"{self.experiments_endpoint}/inputsets/{input_set_id}",
    )

input_set

input_set(input_set_id: str) -> InputSet

Get an input set.

PARAMETER	DESCRIPTION
input_set_id	ID of the input set to retrieve. TYPE: str

RETURNS	DESCRIPTION
InputSet	The requested input set.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> input_set = app.input_set("input-set-123")
>>> print(input_set.name)
'My Input Set'

Source code in nextmv/nextmv/cloud/application/_input_set.py

def input_set(self: "Application", input_set_id: str) -> InputSet:
    """
    Get an input set.

    Parameters
    ----------
    input_set_id : str
        ID of the input set to retrieve.

    Returns
    -------
    InputSet
        The requested input set.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> input_set = app.input_set("input-set-123")
    >>> print(input_set.name)
    'My Input Set'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/inputsets/{input_set_id}",
    )

    return InputSet.from_dict(response.json())

list_input_sets

list_input_sets() -> list[InputSet]

List all input sets.

RETURNS	DESCRIPTION
list[InputSet]	List of all input sets associated with this application.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> input_sets = app.list_input_sets()
>>> for input_set in input_sets:
...     print(input_set.name)
'Input Set 1'
'Input Set 2'

Source code in nextmv/nextmv/cloud/application/_input_set.py

def list_input_sets(self: "Application") -> list[InputSet]:
    """
    List all input sets.

    Returns
    -------
    list[InputSet]
        List of all input sets associated with this application.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> input_sets = app.list_input_sets()
    >>> for input_set in input_sets:
    ...     print(input_set.name)
    'Input Set 1'
    'Input Set 2'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/inputsets",
    )

    return [InputSet.from_dict(input_set) for input_set in response.json()]

new_input_set

new_input_set(
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    end_time: datetime | None = None,
    instance_id: str | None = None,
    maximum_runs: int | None = None,
    run_ids: list[str] | None = None,
    start_time: datetime | None = None,
    inputs: list[ManagedInput] | None = None,
) -> InputSet

Create a new input set. You can create an input set from three different methodologies:

1. Using instance_id, start_time, end_time and maximum_runs. Instance runs will be obtained from the application matching the criteria of dates and maximum number of runs.
2. Using run_ids. The input set will be created using the list of runs specified by the user.
3. Using inputs. The input set will be created using the list of inputs specified by the user. This is useful for creating an input set from a list of inputs that are already available in the application.

PARAMETER	DESCRIPTION
id	ID of the input set, will be generated if not provided. TYPE: str | None DEFAULT: None
name	Name of the input set. If not provided, the ID will be used as the name. TYPE: str | None DEFAULT: None
description	Optional description of the input set. TYPE: str | None DEFAULT: None
end_time	End time of the input set. This is used to filter the runs associated with the input set. TYPE: datetime | None DEFAULT: None
instance_id	ID of the instance to use for the input set. This is used to filter the runs associated with the input set. If not provided, the application's default_instance_id is used. TYPE: str | None DEFAULT: None
maximum_runs	Maximum number of runs to use for the input set. This is used to filter the runs associated with the input set. If not provided, all runs are used. TYPE: int | None DEFAULT: None
run_ids	List of run IDs to use for the input set. TYPE: list[str] | None DEFAULT: None
start_time	Start time of the input set. This is used to filter the runs associated with the input set. TYPE: datetime | None DEFAULT: None
inputs	List of inputs to use for the input set. This is used to create the input set from a list of inputs that are already available in the application. TYPE: list[ManagedInput] | None DEFAULT: None

RETURNS	DESCRIPTION
InputSet	The new input set.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_input_set.py

def new_input_set(
    self: "Application",
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    end_time: datetime | None = None,
    instance_id: str | None = None,
    maximum_runs: int | None = None,
    run_ids: list[str] | None = None,
    start_time: datetime | None = None,
    inputs: list[ManagedInput] | None = None,
) -> InputSet:
    """
    Create a new input set. You can create an input set from three
    different methodologies:

    1. Using `instance_id`, `start_time`, `end_time` and `maximum_runs`.
       Instance runs will be obtained from the application matching the
       criteria of dates and maximum number of runs.
    2. Using `run_ids`. The input set will be created using the list of
       runs specified by the user.
    3. Using `inputs`. The input set will be created using the list of
       inputs specified by the user. This is useful for creating an input
       set from a list of inputs that are already available in the
       application.

    Parameters
    ----------
    id: str | None = None
        ID of the input set, will be generated if not provided.
    name: str | None = None
        Name of the input set. If not provided, the ID will be used as
        the name.
    description: Optional[str]
        Optional description of the input set.
    end_time: Optional[datetime]
        End time of the input set. This is used to filter the runs
        associated with the input set.
    instance_id: Optional[str]
        ID of the instance to use for the input set. This is used to
        filter the runs associated with the input set. If not provided,
        the application's `default_instance_id` is used.
    maximum_runs: Optional[int]
        Maximum number of runs to use for the input set. This is used to
        filter the runs associated with the input set. If not provided,
        all runs are used.
    run_ids: Optional[list[str]]
        List of run IDs to use for the input set.
    start_time: Optional[datetime]
        Start time of the input set. This is used to filter the runs
        associated with the input set.
    inputs: Optional[list[ManagedInput]]
        List of inputs to use for the input set. This is used to create
        the input set from a list of inputs that are already available in
        the application.

    Returns
    -------
    InputSet
        The new input set.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    # Generate ID if not provided
    if id is None or id == "":
        id = safe_id("input-set")

    # Use ID as name if name not provided
    if name is None or name == "":
        name = id

    payload = {
        "id": id,
        "name": name,
    }
    if description is not None:
        payload["description"] = description
    if end_time is not None:
        payload["end_time"] = end_time.isoformat()
    if instance_id is not None:
        payload["instance_id"] = instance_id
    if maximum_runs is not None:
        payload["maximum_runs"] = maximum_runs
    if run_ids is not None:
        payload["run_ids"] = run_ids
    if start_time is not None:
        payload["start_time"] = start_time.isoformat()
    if inputs is not None:
        payload["inputs"] = [input.to_dict() for input in inputs]

    response = self.client.request(
        method="POST",
        endpoint=f"{self.experiments_endpoint}/inputsets",
        payload=payload,
    )

    return InputSet.from_dict(response.json())

update_input_set

update_input_set(
    id: str,
    name: str | None = None,
    description: str | None = None,
    inputs: list[ManagedInput] | None = None,
) -> InputSet

Update an input set.

PARAMETER	DESCRIPTION
id	ID of the input set to update. TYPE: str
name	Optional name of the input set. TYPE: Optional[str] DEFAULT: None
description	Optional description of the input set. TYPE: Optional[str] DEFAULT: None
inputs	List of inputs to use for the input set. This is used to create the input set from a list of inputs that are already available in the application. TYPE: list[ManagedInput] | None DEFAULT: None

RETURNS	DESCRIPTION
Instance	The updated instance.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_input_set.py

def update_input_set(
    self: "Application",
    id: str,
    name: str | None = None,
    description: str | None = None,
    inputs: list[ManagedInput] | None = None,
) -> InputSet:
    """
    Update an input set.

    Parameters
    ----------
    id : str
        ID of the input set to update.
    name : Optional[str], default=None
        Optional name of the input set.
    description : Optional[str], default=None
        Optional description of the input set.
    inputs: Optional[list[ManagedInput]]
        List of inputs to use for the input set. This is used to create
        the input set from a list of inputs that are already available in
        the application.

    Returns
    -------
    Instance
        The updated instance.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    # Get the input set as it currently exsits.
    input_set = self.input_set(id)
    input_set_dict = input_set.to_dict()
    payload = input_set_dict.copy()

    if name is not None:
        payload["name"] = name
    if description is not None:
        payload["description"] = description
    if inputs is not None:
        payload["inputs"] = [input.to_dict() for input in inputs]

    response = self.client.request(
        method="PUT",
        endpoint=f"{self.experiments_endpoint}/inputsets/{id}",
        payload=payload,
    )

    return InputSet.from_dict(response.json())

Instance Management

_instance

Application mixin for managing app instances.

ApplicationInstanceMixin

Mixin class for managing app instances within an application.

delete_instance

delete_instance(instance_id: str) -> None

Delete an instance.

Permanently removes the specified instance from the application.

PARAMETER	DESCRIPTION
instance_id	ID of the instance to delete. TYPE: str

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> app.delete_instance("prod-instance")  # Permanently deletes the instance

Source code in nextmv/nextmv/cloud/application/_instance.py

def delete_instance(self: "Application", instance_id: str) -> None:
    """
    Delete an instance.

    Permanently removes the specified instance from the application.

    Parameters
    ----------
    instance_id : str
        ID of the instance to delete.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> app.delete_instance("prod-instance")  # Permanently deletes the instance
    """

    _ = self.client.request(
        method="DELETE",
        endpoint=f"{self.endpoint}/instances/{instance_id}",
    )

instance

instance(instance_id: str) -> Instance

Get an instance.

PARAMETER	DESCRIPTION
instance_id	ID of the instance to retrieve. TYPE: str

RETURNS	DESCRIPTION
Instance	The requested instance details.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> instance = app.instance("instance-123")
>>> print(instance.name)
'Production Instance'

Source code in nextmv/nextmv/cloud/application/_instance.py

def instance(self: "Application", instance_id: str) -> Instance:
    """
    Get an instance.

    Parameters
    ----------
    instance_id : str
        ID of the instance to retrieve.

    Returns
    -------
    Instance
        The requested instance details.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> instance = app.instance("instance-123")
    >>> print(instance.name)
    'Production Instance'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.endpoint}/instances/{instance_id}",
    )

    return Instance.from_dict(response.json())

instance_exists

instance_exists(instance_id: str) -> bool

Check if an instance exists.

PARAMETER	DESCRIPTION
instance_id	ID of the instance to check. TYPE: str

RETURNS	DESCRIPTION
bool	True if the instance exists, False otherwise.

Examples:

>>> app.instance_exists("instance-123")
True

Source code in nextmv/nextmv/cloud/application/_instance.py

def instance_exists(self: "Application", instance_id: str) -> bool:
    """
    Check if an instance exists.

    Parameters
    ----------
    instance_id : str
        ID of the instance to check.

    Returns
    -------
    bool
        True if the instance exists, False otherwise.

    Examples
    --------
    >>> app.instance_exists("instance-123")
    True
    """

    try:
        self.instance(instance_id=instance_id)
        return True
    except requests.HTTPError as e:
        if _is_not_exist_error(e):
            return False
        raise e

list_instances

list_instances() -> list[Instance]

List all instances.

RETURNS	DESCRIPTION
list[Instance]	List of all instances associated with this application.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> instances = app.list_instances()
>>> for instance in instances:
...     print(instance.name)
'Development Instance'
'Production Instance'

Source code in nextmv/nextmv/cloud/application/_instance.py

def list_instances(self: "Application") -> list[Instance]:
    """
    List all instances.

    Returns
    -------
    list[Instance]
        List of all instances associated with this application.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> instances = app.list_instances()
    >>> for instance in instances:
    ...     print(instance.name)
    'Development Instance'
    'Production Instance'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.endpoint}/instances",
    )

    return [Instance.from_dict(instance) for instance in response.json()]

new_instance

new_instance(
    version_id: str,
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    configuration: InstanceConfiguration | None = None,
    exist_ok: bool = False,
) -> Instance

Create a new instance and associate it with a version.

This method creates a new instance associated with a specific version of the application. Instances are configurations of an application version that can be executed.

PARAMETER	DESCRIPTION
version_id	ID of the version to associate the instance with. TYPE: str
id	ID of the instance. Will be generated if not provided. TYPE: str | None DEFAULT: None
name	Name of the instance. Will be generated if not provided. TYPE: str | None DEFAULT: None
description	Description of the instance. TYPE: Optional[str] DEFAULT: None
configuration	Configuration to use for the instance. This can include resources, timeouts, and other execution parameters. TYPE: Optional[InstanceConfiguration] DEFAULT: None
exist_ok	If True and an instance with the same ID already exists, return the existing instance instead of creating a new one. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
Instance	The newly created (or existing) instance.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.
ValueError	If exist_ok is True and id is None.

Examples:

>>> # Create a new instance for a specific version
>>> instance = app.new_instance(
...     version_id="version-123",
...     id="prod-instance",
...     name="Production Instance",
...     description="Instance for production use"
... )
>>> print(instance.name)
'Production Instance'

Source code in nextmv/nextmv/cloud/application/_instance.py

def new_instance(
    self: "Application",
    version_id: str,
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    configuration: InstanceConfiguration | None = None,
    exist_ok: bool = False,
) -> Instance:
    """
    Create a new instance and associate it with a version.

    This method creates a new instance associated with a specific version
    of the application. Instances are configurations of an application
    version that can be executed.

    Parameters
    ----------
    version_id : str
        ID of the version to associate the instance with.
    id : str | None, default=None
        ID of the instance. Will be generated if not provided.
    name : str | None, default=None
        Name of the instance. Will be generated if not provided.
    description : Optional[str], default=None
        Description of the instance.
    configuration : Optional[InstanceConfiguration], default=None
        Configuration to use for the instance. This can include resources,
        timeouts, and other execution parameters.
    exist_ok : bool, default=False
        If True and an instance with the same ID already exists,
        return the existing instance instead of creating a new one.

    Returns
    -------
    Instance
        The newly created (or existing) instance.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    ValueError
        If exist_ok is True and id is None.

    Examples
    --------
    >>> # Create a new instance for a specific version
    >>> instance = app.new_instance(
    ...     version_id="version-123",
    ...     id="prod-instance",
    ...     name="Production Instance",
    ...     description="Instance for production use"
    ... )
    >>> print(instance.name)
    'Production Instance'
    """

    if exist_ok and (id is None or id == ""):
        raise ValueError("If exist_ok is True, id must be provided")

    if exist_ok and self.instance_exists(instance_id=id):
        return self.instance(instance_id=id)

    if id is None or id == "":
        id = safe_id(prefix="instance")
    if name is None or name == "":
        name = id

    payload = {
        "id": id,
        "name": name,
        "version_id": version_id,
    }

    if description is not None:
        payload["description"] = description
    if configuration is not None:
        payload["configuration"] = configuration.to_dict()

    response = self.client.request(
        method="POST",
        endpoint=f"{self.endpoint}/instances",
        payload=payload,
    )

    return Instance.from_dict(response.json())

update_instance

update_instance(
    id: str,
    name: str | None = None,
    version_id: str | None = None,
    description: str | None = None,
    configuration: InstanceConfiguration
    | dict[str, Any]
    | None = None,
    locked: bool | None = None,
) -> Instance

Update an instance.

PARAMETER	DESCRIPTION
id	ID of the instance to update. TYPE: str
name	Optional name of the instance. TYPE: Optional[str] DEFAULT: None
version_id	Optional ID of the version to associate the instance with. TYPE: Optional[str] DEFAULT: None
description	Optional description of the instance. TYPE: Optional[str] DEFAULT: None
configuration	Optional configuration to use for the instance. TYPE: Optional[InstanceConfiguration | dict[str, Any]] DEFAULT: None
locked	Optional locked status of the instance. If True, the instance will be locked. If False, the instance will be unlocked. If None, the locked status will not be updated. TYPE: Optional[bool] DEFAULT: None

RETURNS	DESCRIPTION
Instance	The updated instance.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_instance.py

def update_instance(
    self: "Application",
    id: str,
    name: str | None = None,
    version_id: str | None = None,
    description: str | None = None,
    configuration: InstanceConfiguration | dict[str, Any] | None = None,
    locked: bool | None = None,
) -> Instance:
    """
    Update an instance.

    Parameters
    ----------
    id : str
        ID of the instance to update.
    name : Optional[str], default=None
        Optional name of the instance.
    version_id : Optional[str], default=None
        Optional ID of the version to associate the instance with.
    description : Optional[str], default=None
        Optional description of the instance.
    configuration : Optional[InstanceConfiguration | dict[str, Any]], default=None
        Optional configuration to use for the instance.
    locked : Optional[bool], default=None
        Optional locked status of the instance. If True, the instance will
        be locked. If False, the instance will be unlocked. If None, the
        locked status will not be updated.

    Returns
    -------
    Instance
        The updated instance.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    # Get the instance as it currently exsits.
    instance = self.instance(id)
    instance_dict = instance.to_dict()
    payload = instance_dict.copy()

    if name is not None:
        payload["name"] = name
    if version_id is not None:
        payload["version_id"] = version_id
    if description is not None:
        payload["description"] = description
    if configuration is not None:
        if isinstance(configuration, dict):
            config_dict = configuration
        elif isinstance(configuration, InstanceConfiguration):
            config_dict = configuration.to_dict()
        else:
            raise TypeError("configuration must be either a dict or InstanceConfiguration object")

        payload["configuration"] = config_dict

    response = self.client.request(
        method="PUT",
        endpoint=f"{self.endpoint}/instances/{id}",
        payload=payload,
    )

    if locked is None:
        return Instance.from_dict(response.json())

    lock_payload = {"locked": locked}
    lock_resp = self.client.request(
        method="PUT",
        endpoint=f"{self.endpoint}/instances/{id}/lock",
        payload=lock_payload,
    )

    return Instance.from_dict(lock_resp.json())

Managed Inputs

_managed_input

Application mixin for handling app managed inputs.

ApplicationManagedInputMixin

Mixin class for handling app managed inputs within an application.

delete_managed_input

delete_managed_input(managed_input_id: str) -> None

Delete a managed input.

Permanently removes the specified managed input from the application.

PARAMETER	DESCRIPTION
managed_input_id	ID of the managed input to delete. TYPE: str

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> app.delete_managed_input("inp_123456789")  # Permanently deletes the managed input

Source code in nextmv/nextmv/cloud/application/_managed_input.py

def delete_managed_input(self: "Application", managed_input_id: str) -> None:
    """
    Delete a managed input.

    Permanently removes the specified managed input from the application.

    Parameters
    ----------
    managed_input_id : str
        ID of the managed input to delete.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> app.delete_managed_input("inp_123456789")  # Permanently deletes the managed input
    """

    _ = self.client.request(
        method="DELETE",
        endpoint=f"{self.endpoint}/inputs/{managed_input_id}",
    )

list_managed_inputs

list_managed_inputs() -> list[ManagedInput]

List all managed inputs.

RETURNS	DESCRIPTION
list[ManagedInput]	List of managed inputs.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_managed_input.py

def list_managed_inputs(self: "Application") -> list[ManagedInput]:
    """
    List all managed inputs.

    Returns
    -------
    list[ManagedInput]
        List of managed inputs.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.endpoint}/inputs",
    )

    return [ManagedInput.from_dict(managed_input) for managed_input in response.json()]

managed_input

managed_input(managed_input_id: str) -> ManagedInput

Get a managed input.

PARAMETER	DESCRIPTION
managed_input_id	ID of the managed input. TYPE: str

RETURNS	DESCRIPTION
ManagedInput	The managed input.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_managed_input.py

def managed_input(self: "Application", managed_input_id: str) -> ManagedInput:
    """
    Get a managed input.

    Parameters
    ----------
    managed_input_id: str
        ID of the managed input.

    Returns
    -------
    ManagedInput
        The managed input.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.endpoint}/inputs/{managed_input_id}",
    )

    return ManagedInput.from_dict(response.json())

new_managed_input

new_managed_input(
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    upload_id: str | None = None,
    run_id: str | None = None,
    format: Format | dict[str, Any] | None = None,
) -> ManagedInput

Create a new managed input. There are two methods for creating a managed input:

1. Specifying the upload_id parameter. You may use the upload_url method to obtain the upload ID and the upload_data method to upload the data to it.
2. Specifying the run_id parameter. The managed input will be created from the run specified by the run_id parameter.

Either the upload_id or the run_id parameter must be specified.

PARAMETER	DESCRIPTION
id	ID of the managed input. Will be generated if not provided. TYPE: str | None DEFAULT: None
name	Name of the managed input. Will be generated if not provided. TYPE: str | None DEFAULT: None
description	Optional description of the managed input. TYPE: str | None DEFAULT: None
upload_id	ID of the upload to use for the managed input. TYPE: str | None DEFAULT: None
run_id	ID of the run to use for the managed input. TYPE: str | None DEFAULT: None
format	Format of the managed input. Default will be formatted as JSON. TYPE: Format | dict[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
ManagedInput	The new managed input.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.
ValueError	If neither the upload_id nor the run_id parameter is specified.

Source code in nextmv/nextmv/cloud/application/_managed_input.py

def new_managed_input(
    self: "Application",
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    upload_id: str | None = None,
    run_id: str | None = None,
    format: Format | dict[str, Any] | None = None,
) -> ManagedInput:
    """
    Create a new managed input. There are two methods for creating a
    managed input:

    1. Specifying the `upload_id` parameter. You may use the `upload_url`
       method to obtain the upload ID and the `upload_data` method
       to upload the data to it.
    2. Specifying the `run_id` parameter. The managed input will be
       created from the run specified by the `run_id` parameter.

    Either the `upload_id` or the `run_id` parameter must be specified.

    Parameters
    ----------
    id: Optional[str], default=None
        ID of the managed input. Will be generated if not provided.
    name: Optional[str], default=None
        Name of the managed input. Will be generated if not provided.
    description: Optional[str], default=None
        Optional description of the managed input.
    upload_id: Optional[str], default=None
        ID of the upload to use for the managed input.
    run_id: Optional[str], default=None
        ID of the run to use for the managed input.
    format: Optional[Format], default=None
        Format of the managed input. Default will be formatted as `JSON`.

    Returns
    -------
    ManagedInput
        The new managed input.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    ValueError
        If neither the `upload_id` nor the `run_id` parameter is
        specified.
    """

    if upload_id is None and run_id is None:
        raise ValueError("Either upload_id or run_id must be specified")

    if id is None or id == "":
        id = safe_id(prefix="managed-input")
    if name is None or name == "":
        name = id

    payload = {
        "id": id,
        "name": name,
    }

    if description is not None:
        payload["description"] = description
    if upload_id is not None:
        payload["upload_id"] = upload_id
    if run_id is not None:
        payload["run_id"] = run_id

    if format is not None:
        payload["format"] = format.to_dict() if isinstance(format, Format) else format
    else:
        payload["format"] = Format(
            format_input=FormatInput(input_type=InputFormat.JSON),
            format_output=FormatOutput(output_type=OutputFormat.JSON),
        ).to_dict()

    response = self.client.request(
        method="POST",
        endpoint=f"{self.endpoint}/inputs",
        payload=payload,
    )

    return ManagedInput.from_dict(response.json())

update_managed_input

update_managed_input(
    managed_input_id: str,
    name: str | None = None,
    description: str | None = None,
) -> ManagedInput

Update a managed input.

PARAMETER	DESCRIPTION
managed_input_id	ID of the managed input to update. TYPE: str
name	Optional new name for the managed input. TYPE: Optional[str] DEFAULT: None
description	Optional new description for the managed input. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
ManagedInput	The updated managed input.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_managed_input.py

def update_managed_input(
    self: "Application",
    managed_input_id: str,
    name: str | None = None,
    description: str | None = None,
) -> ManagedInput:
    """
    Update a managed input.

    Parameters
    ----------
    managed_input_id : str
        ID of the managed input to update.
    name : Optional[str], default=None
        Optional new name for the managed input.
    description : Optional[str], default=None
        Optional new description for the managed input.

    Returns
    -------
    ManagedInput
        The updated managed input.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    managed_input = self.managed_input(managed_input_id)
    managed_input_dict = managed_input.to_dict()
    payload = managed_input_dict.copy()

    if name is not None:
        payload["name"] = name
    if description is not None:
        payload["description"] = description

    response = self.client.request(
        method="PUT",
        endpoint=f"{self.endpoint}/inputs/{managed_input_id}",
        payload=payload,
    )

    return ManagedInput.from_dict(response.json())

Run Management

_run

Application mixin for managing app runs.

ApplicationRunMixin

Mixin class for managing app runs within an application.

cancel_run

cancel_run(run_id: str) -> None

Cancel a run.

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

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> app.cancel_run("run-456")

Source code in nextmv/nextmv/cloud/application/_run.py

def cancel_run(self: "Application", run_id: str) -> None:
    """
    Cancel a run.

    Parameters
    ----------
    run_id : str
        ID of the run to cancel.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> app.cancel_run("run-456")
    """

    _ = self.client.request(
        method="PATCH",
        endpoint=f"{self.endpoint}/runs/{run_id}/cancel",
    )

download_asset_content

download_asset_content(
    asset: RunAsset,
    destination: str | Path | BytesIO | None = None,
) -> Any | None

Downloads an asset's content to a specified destination.

PARAMETER	DESCRIPTION
asset	The asset to be downloaded. TYPE: RunAsset
destination	The destination where the asset will be saved. This can be a file path (as a string or pathlib.Path) or an io.BytesIO object. If None, the asset content will not be saved to a file, but returned immediately. If the asset type is JSON, the content will be returned as a dict. TYPE: str | Path | BytesIO | None DEFAULT: None

RETURNS	DESCRIPTION
Any or None	If destination is None, returns the asset content: for JSON assets, a dict parsed from the JSON response; for other asset types, the raw bytes content. If destination is provided, the content is written to the given destination and the method returns None.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> assets = app.list_assets("run-123")
>>> asset = assets[0]  # Assume we want to download the first asset
>>> # Download to a file path
>>> app.download_asset_content(asset, "polygons.geojson")
>>> # Download to an in-memory bytes buffer
>>> import io
>>> buffer = io.BytesIO()
>>> app.download_asset_content(asset, buffer)
>>> # Download and get content directly (for JSON assets)
>>> content = app.download_asset_content(asset)
>>> print(content)
{'type': 'FeatureCollection', 'features': [...]}

Source code in nextmv/nextmv/cloud/application/_run.py

def download_asset_content(
    self: "Application",
    asset: RunAsset,
    destination: str | pathlib.Path | io.BytesIO | None = None,
) -> Any | None:
    """
    Downloads an asset's content to a specified destination.

    Parameters
    ----------
    asset : RunAsset
        The asset to be downloaded.
    destination : str | pathlib.Path | io.BytesIO | None
        The destination where the asset will be saved. This can be a file path
        (as a string or pathlib.Path) or an io.BytesIO object. If None, the asset
        content will not be saved to a file, but returned immediately. If the asset
        type is JSON, the content will be returned as a dict.

    Returns
    -------
    Any or None
        If `destination` is None, returns the asset content: for JSON assets, a
        `dict` parsed from the JSON response; for other asset types, the raw
        `bytes` content. If `destination` is provided, the content is written
        to the given destination and the method returns `None`.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> assets = app.list_assets("run-123")
    >>> asset = assets[0]  # Assume we want to download the first asset
    >>> # Download to a file path
    >>> app.download_asset_content(asset, "polygons.geojson")
    >>> # Download to an in-memory bytes buffer
    >>> import io
    >>> buffer = io.BytesIO()
    >>> app.download_asset_content(asset, buffer)
    >>> # Download and get content directly (for JSON assets)
    >>> content = app.download_asset_content(asset)
    >>> print(content)
    {'type': 'FeatureCollection', 'features': [...]}
    """
    # First, get the download_url for the asset.
    download_url_response = self.client.request(
        method="GET",
        endpoint=f"{self.endpoint}/runs/{asset.run_id}/assets/{asset.id}",
    ).json()
    download_url = download_url_response["download_url"]
    asset_type = download_url_response.get("type", "json")

    # Now, download the asset content using the download_url.
    download_response = self.client.request(
        method="GET",
        endpoint=download_url,
        headers={"Content-Type": "application/json" if asset_type == "json" else "application/octet-stream"},
    )

    # Save the content to the specified destination.
    if destination is None:
        if asset_type == "json":
            return download_response.json()
        return download_response.content
    elif isinstance(destination, io.BytesIO):
        destination.write(download_response.content)
        return None
    else:
        with open(destination, "wb") as file:
            file.write(download_response.content)
        return None

list_assets

list_assets(run_id: str) -> list[RunAsset]

List the assets of a run.

Retrieves a list of assets associated with a specific run. This method ONLY returns the asset metadata, the content needs to be fetched via the download_asset_content method.

PARAMETER	DESCRIPTION
run_id	ID of the run to list assets for. TYPE: str

RETURNS	DESCRIPTION
list[RunAsset]	List of assets associated with the run.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> assets = app.list_assets("run-123")
>>> for asset in assets:
...     print(asset.id, asset.name)
b459daa6-1c13-48c6-b4c3-a262ea94cd04 clustering_polygons
a1234567-89ab-cdef-0123-456789abcdef histogram

Source code in nextmv/nextmv/cloud/application/_run.py

def list_assets(self: "Application", run_id: str) -> list[RunAsset]:
    """
    List the assets of a run.

    Retrieves a list of assets associated with a specific run. This method ONLY
    returns the asset metadata, the content needs to be fetched via the
    `download_asset_content` method.

    Parameters
    ----------
    run_id : str
        ID of the run to list assets for.

    Returns
    -------
    list[RunAsset]
        List of assets associated with the run.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> assets = app.list_assets("run-123")
    >>> for asset in assets:
    ...     print(asset.id, asset.name)
    b459daa6-1c13-48c6-b4c3-a262ea94cd04 clustering_polygons
    a1234567-89ab-cdef-0123-456789abcdef histogram
    """
    response = self.client.request(
        method="GET",
        endpoint=f"{self.endpoint}/runs/{run_id}/assets",
    )
    assets_data = response.json().get("items", [])
    for asset_data in assets_data:
        asset_data["run_id"] = run_id

    return [RunAsset.from_dict(asset) for asset in assets_data]

list_runs

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

List all runs.

You can use the optional status parameter to filter runs by their status. Is not provided, all runs are returned.

PARAMETER	DESCRIPTION
status	Optional status to filter runs by. TYPE: StatusV2 | None DEFAULT: None

RETURNS	DESCRIPTION
list[Run]	List of runs.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_run.py

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

    You can use the optional `status` parameter to filter runs by their
    status. Is not provided, all runs are returned.

    Parameters
    ----------
    status : StatusV2 | None
        Optional status to filter runs by.

    Returns
    -------
    list[Run]
        List of runs.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.endpoint}/runs",
    )

    runs = []
    for resp_run in response.json().get("runs", []):
        run = Run.from_dict(resp_run)
        if status is None:
            runs.append(run)
            continue

        if run.status_v2 == status:
            runs.append(run)

    return runs

new_run

new_run(
    input: Input | dict[str, Any] | BaseModel | str = None,
    instance_id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    upload_id: str | None = None,
    options: Options | dict[str, str] | None = None,
    configuration: RunConfiguration
    | dict[str, Any]
    | None = None,
    batch_experiment_id: str | None = None,
    external_result: ExternalRunResult
    | dict[str, Any]
    | None = None,
    json_configurations: dict[str, Any] | None = None,
    input_dir_path: str | None = None,
) -> str

Submit an input to start a new run of the application. Returns the run_id of the submitted run.

PARAMETER	DESCRIPTION
input	Input to use for the run. This can be a nextmv.Input object, dict, BaseModel or str. If nextmv.Input is used, and the input_format is either nextmv.InputFormat.JSON or nextmv.InputFormat.TEXT, then the input data is extracted from the .data property. If you want to work with nextmv.InputFormat.CSV_ARCHIVE or nextmv.InputFormat.MULTI_FILE, you should use the input_dir_path argument instead. This argument takes precedence over the input. If input_dir_path is specified, this function looks for files in that directory and tars them, to later be uploaded using the upload_data method. If both the input_dir_path and input arguments are provided, the input is ignored. When input_dir_path is specified, the configuration argument must also be provided. More specifically, the RunConfiguration.format.format_input.input_type parameter dictates what kind of input is being submitted to the Nextmv Cloud. Make sure that this parameter is specified when working with the following input formats: nextmv.InputFormat.CSV_ARCHIVE nextmv.InputFormat.MULTI_FILE When working with JSON or text data, use the input argument directly. In general, if an input is too large, it will be uploaded with the upload_data method. TYPE: Input | dict[str, Any] | BaseModel | str DEFAULT: None
instance_id	ID of the instance to use for the run. If not provided, the default instance ID associated to the Class (default_instance_id) is used. TYPE: str | None DEFAULT: None
name	Name of the run. TYPE: str | None DEFAULT: None
description	Description of the run. TYPE: str | None DEFAULT: None
upload_id	ID to use when running a large input. If the input exceeds the maximum allowed size, then it is uploaded and the corresponding upload_id is used. 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
batch_experiment_id	ID of a batch experiment to associate the run with. This is used when the run is part of a batch experiment. TYPE: str | None DEFAULT: None
external_result	External result to use for the run. This can be a nextmv.ExternalRunResult object or a dict. If the object is used, then the .to_dict() method is applied to extract the configuration. This is used when the run is an external run. We suggest that instead of specifying this parameter, you use the track_run method of the class. TYPE: ExternalRunResult | 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. If specified, the function will package the files in the directory into a tar file and upload it as a large input. 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 run that was submitted.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.
ValueError	If the input is of type nextmv.Input and the .input_formatis notJSON. If the finaloptionsare not of typedict[str,str]`.

Source code in nextmv/nextmv/cloud/application/_run.py

def new_run(  # noqa: C901 # Refactor this function at some point.
    self: "Application",
    input: Input | dict[str, Any] | BaseModel | str = None,
    instance_id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    upload_id: str | None = None,
    options: Options | dict[str, str] | None = None,
    configuration: RunConfiguration | dict[str, Any] | None = None,
    batch_experiment_id: str | None = None,
    external_result: ExternalRunResult | dict[str, Any] | None = None,
    json_configurations: dict[str, Any] | None = None,
    input_dir_path: str | None = None,
) -> str:
    """
    Submit an input to start a new run of the application. Returns the
    `run_id` of the submitted run.

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

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

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

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

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

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

        In general, if an input is too large, it will be uploaded with the
        `upload_data` method.
    instance_id: Optional[str]
        ID of the instance to use for the run. If not provided, the default
        instance ID associated to the Class (`default_instance_id`) is
        used.
    name: Optional[str]
        Name of the run.
    description: Optional[str]
        Description of the run.
    upload_id: Optional[str]
        ID to use when running a large input. If the `input` exceeds the
        maximum allowed size, then it is uploaded and the corresponding
        `upload_id` is used.
    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.
    batch_experiment_id: Optional[str]
        ID of a batch experiment to associate the run with. This is used
        when the run is part of a batch experiment.
    external_result: Optional[Union[ExternalRunResult, dict[str, Any]]]
        External result to use for the run. This can be a
        `nextmv.ExternalRunResult` object or a dict. If the object is used,
        then the `.to_dict()` method is applied to extract the
        configuration. This is used when the run is an external run. We
        suggest that instead of specifying this parameter, you use the
        `track_run` method of the class.
    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. If specified, the
        function will package the files in the directory into a tar file
        and upload it as a large input. 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 run that was submitted.

    Raises
    ----------
    requests.HTTPError
        If the response status code is not 2xx.
    ValueError
        If the `input` is of type `nextmv.Input` and the .input_format` is
        not `JSON`. If the final `options` are not of type `dict[str,str]`.
    """

    tar_file = ""
    if input_dir_path is not None and input_dir_path != "":
        if not os.path.exists(input_dir_path):
            raise ValueError(f"Directory {input_dir_path} does not exist.")

        if not os.path.isdir(input_dir_path):
            raise ValueError(f"Path {input_dir_path} is not a directory.")

        tar_file = self._package_inputs(input_dir_path)

    input_data = self.__extract_input_data(input)

    input_size = 0
    if input_data is not None:
        input_size = get_size(input_data)

    upload_id_used = upload_id is not None
    if self.__upload_url_required(upload_id_used, input_size, tar_file, input):
        upload_url = self.upload_url()
        self.upload_data(data=input_data, upload_url=upload_url, tar_file=tar_file)
        upload_id = upload_url.upload_id
        upload_id_used = True

    options_dict = self.__extract_options_dict(options, json_configurations)

    # Builds the payload progressively based on the different arguments
    # that must be provided.
    payload = {}
    if upload_id_used:
        payload["upload_id"] = upload_id
    else:
        payload["input"] = input_data

    if name is not None:
        payload["name"] = name
    if description is not None:
        payload["description"] = description
    if len(options_dict) > 0:
        for k, v in options_dict.items():
            if not isinstance(v, str):
                raise ValueError(f"options must be dict[str,str], option {k} has type {type(v)} instead.")
        payload["options"] = options_dict

    configuration_dict = self.__extract_run_config(input, configuration, input_dir_path)
    payload["configuration"] = configuration_dict

    if batch_experiment_id is not None:
        payload["batch_experiment_id"] = batch_experiment_id
    if external_result is not None:
        external_dict = (
            external_result.to_dict() if isinstance(external_result, ExternalRunResult) else external_result
        )
        payload["result"] = external_dict

    query_params = {}
    if instance_id is not None or self.default_instance_id is not None:
        query_params["instance_id"] = instance_id if instance_id is not None else self.default_instance_id

    response = self.client.request(
        method="POST",
        endpoint=f"{self.endpoint}/runs",
        payload=payload,
        query_params=query_params,
        json_configurations=json_configurations,
    )

    return response.json()["run_id"]

new_run_with_result

new_run_with_result(
    input: Input | dict[str, Any] | BaseModel | str = None,
    instance_id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    upload_id: str | None = None,
    run_options: Options | dict[str, str] | None = None,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
    configuration: RunConfiguration
    | dict[str, Any]
    | None = None,
    batch_experiment_id: str | None = None,
    external_result: ExternalRunResult
    | 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 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 run succeeded.

PARAMETER	DESCRIPTION
input	Input to use for the run. This can be a nextmv.Input object, dict, BaseModel or str. If nextmv.Input is used, and the input_format is either nextmv.InputFormat.JSON or nextmv.InputFormat.TEXT, then the input data is extracted from the .data property. If you want to work with nextmv.InputFormat.CSV_ARCHIVE or nextmv.InputFormat.MULTI_FILE, you should use the input_dir_path argument instead. This argument takes precedence over the input. If input_dir_path is specified, this function looks for files in that directory and tars them, to later be uploaded using the upload_data method. If both the input_dir_path and input arguments are provided, the input is ignored. When input_dir_path is specified, the configuration argument must also be provided. More specifically, the RunConfiguration.format.format_input.input_type parameter dictates what kind of input is being submitted to the Nextmv Cloud. Make sure that this parameter is specified when working with the following input formats: nextmv.InputFormat.CSV_ARCHIVE nextmv.InputFormat.MULTI_FILE When working with JSON or text data, use the input argument directly. In general, if an input is too large, it will be uploaded with the upload_data method. TYPE: Input | dict[str, Any] | BaseModel | str DEFAULT: None
instance_id	ID of the instance to use for the run. If not provided, the default instance ID associated to the Class (default_instance_id) is used. TYPE: str | None DEFAULT: None
name	Name of the run. TYPE: str | None DEFAULT: None
description	Description of the run. TYPE: str | None DEFAULT: None
upload_id	ID to use when running a large input. If the input exceeds the maximum allowed size, then it is uploaded and the corresponding upload_id is used. 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. This is a convenience method that combines the new_run and run_result_with_polling methods, applying polling logic to check when the run succeeded. 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
batch_experiment_id	ID of a batch experiment to associate the run with. This is used when the run is part of a batch experiment. TYPE: str | None DEFAULT: None
external_result	External result to use for the run. This can be a cloud.ExternalRunResult object or a dict. If the object is used, then the .to_dict() method is applied to extract the configuration. This is used when the run is an external run. We suggest that instead of specifying this parameter, you use the track_run_with_result method of the class. TYPE: ExternalRunResult | 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. If specified, the function will package the files in the directory into a tar file and upload it as a large input. 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.

RAISES	DESCRIPTION
ValueError	If the input is of type nextmv.Input and the .input_format is not JSON. If the final options are not of type dict[str,str].
HTTPError	If the response status code is not 2xx.
TimeoutError	If the run does not succeed after the polling strategy is exhausted based on time duration.
RuntimeError	If the run does not succeed after the polling strategy is exhausted based on number of tries.

Source code in nextmv/nextmv/cloud/application/_run.py

def new_run_with_result(
    self: "Application",
    input: Input | dict[str, Any] | BaseModel | str = None,
    instance_id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    upload_id: str | None = None,
    run_options: Options | dict[str, str] | None = None,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
    configuration: RunConfiguration | dict[str, Any] | None = None,
    batch_experiment_id: str | None = None,
    external_result: ExternalRunResult | 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 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 run succeeded.

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

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

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

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

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

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

        In general, if an input is too large, it will be uploaded with the
        `upload_data` method.
    instance_id: Optional[str]
        ID of the instance to use for the run. If not provided, the default
        instance ID associated to the Class (`default_instance_id`) is
        used.
    name: Optional[str]
        Name of the run.
    description: Optional[str]
        Description of the run.
    upload_id: Optional[str]
        ID to use when running a large input. If the `input` exceeds the
        maximum allowed size, then it is uploaded and the corresponding
        `upload_id` is used.
    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
        Options to use when polling for the run result. This is a
        convenience method that combines the `new_run` and
        `run_result_with_polling` methods, applying polling logic to check
        when the run succeeded.
    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.
    batch_experiment_id: Optional[str]
        ID of a batch experiment to associate the run with. This is used
        when the run is part of a batch experiment.
    external_result: Optional[Union[ExternalRunResult, dict[str, Any]]] = None
        External result to use for the run. This can be a
        `cloud.ExternalRunResult` object or a dict. If the object is used,
        then the `.to_dict()` method is applied to extract the
        configuration. This is used when the run is an external run. We
        suggest that instead of specifying this parameter, you use the
        `track_run_with_result` method of the class.
    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. If specified, the
        function will package the files in the directory into a tar file
        and upload it as a large input. 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.

    Raises
    ----------
    ValueError
        If the `input` is of type `nextmv.Input` and the `.input_format` is
        not `JSON`. If the final `options` are not of type `dict[str,str]`.
    requests.HTTPError
        If the response status code is not 2xx.
    TimeoutError
        If the run does not succeed after the polling strategy is exhausted
        based on time duration.
    RuntimeError
        If the run does not succeed after the polling strategy is exhausted
        based on number of tries.
    """

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

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

run_input

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

Get the input of a run.

Retrieves the input data that was used for a specific run. This method handles both small and large inputs automatically - if the input size exceeds the maximum allowed size, it will fetch the input from a download URL. If the content format of the run is csv-archive or multi-file, then the output_dir_path parameter must be specified.

PARAMETER	DESCRIPTION
run_id	ID of the run to retrieve the 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] | None	Input data of the run as a dictionary. If the input format is non-JSON (e.g., csv-archive or multi-file), the method returns None after saving the input files to the specified output_dir_path.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> input_data = app.run_input("run-123")
>>> print(input_data)
{'locations': [...], 'vehicles': [...]}

Source code in nextmv/nextmv/cloud/application/_run.py

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

    Retrieves the input data that was used for a specific run. This method
    handles both small and large inputs automatically - if the input size
    exceeds the maximum allowed size, it will fetch the input from a
    download URL. If the content format of the run is `csv-archive` or
    `multi-file`, then the `output_dir_path` parameter must be specified.

    Parameters
    ----------
    run_id : str
        ID of the run to retrieve the 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] | None
        Input data of the run as a dictionary. If the input format is
        non-JSON (e.g., csv-archive or multi-file), the method returns None
        after saving the input files to the specified `output_dir_path`.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> input_data = app.run_input("run-123")
    >>> print(input_data)
    {'locations': [...], 'vehicles': [...]}
    """
    run_information = self.run_metadata(run_id=run_id)

    query_params = None
    large = False
    if (
        run_information.metadata.input_size > _MAX_RUN_SIZE
        or run_information.metadata.format.format_input.input_type
        in {InputFormat.CSV_ARCHIVE, InputFormat.MULTI_FILE}
    ):
        query_params = {"format": "url"}
        large = True

    response = self.client.request(
        method="GET",
        endpoint=f"{self.endpoint}/runs/{run_id}/input",
        query_params=query_params,
    )
    if not large:
        return response.json()

    download_url = DownloadURL.from_dict(response.json())
    download_response = self.client.request(
        method="GET",
        endpoint=download_url.url,
        headers={"Content-Type": "application/json"},
    )

    # See whether we can return the input directly or need to save to the given
    # directory
    if run_information.metadata.format.format_input.input_type != OutputFormat.JSON:
        if not output_dir_path or output_dir_path == "":
            raise ValueError(
                "If the input format is not JSON, an output_dir_path must be provided.",
            )
        if not os.path.exists(output_dir_path):
            os.makedirs(output_dir_path, exist_ok=True)

        # Save .tar.gz file to a temp directory and extract contents to output_dir_path
        with tempfile.TemporaryDirectory() as tmpdirname:
            temp_tar_path = os.path.join(tmpdirname, f"{run_id}.tar.gz")
            with open(temp_tar_path, "wb") as f:
                f.write(download_response.content)
            shutil.unpack_archive(temp_tar_path, output_dir_path)

        return

    # JSON input can be returned directly.
    return download_response.json()

run_logs

run_logs(run_id: str) -> RunLog

Get the logs of a run.

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

RETURNS	DESCRIPTION
RunLog	Logs of the run.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> logs = app.run_logs("run-123")
>>> print(logs.stderr)
'Warning: resource usage exceeded'

Source code in nextmv/nextmv/cloud/application/_run.py

def run_logs(self: "Application", run_id: str) -> RunLog:
    """
    Get the logs of a run.

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

    Returns
    -------
    RunLog
        Logs of the run.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> logs = app.run_logs("run-123")
    >>> print(logs.stderr)
    'Warning: resource usage exceeded'
    """
    response = self.client.request(
        method="GET",
        endpoint=f"{self.endpoint}/runs/{run_id}/logs",
    )

    return RunLog.from_dict(response.json())

run_logs_with_polling

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

Get the logs of a 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 sorted by timestamp. 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

RETURNS	DESCRIPTION
list[TimestampedRunLog]	List of timestamped logs of the run.

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 logs with polling
>>> logs = app.run_logs_with_polling("run-123", polling_opts)
>>> for log in logs:
...     print(f"[{log.timestamp}] {log.log}")
[2024-01-01T12:00:00Z] Starting optimization...
[2024-01-01T12:00:05Z] Found initial solution
...

Source code in nextmv/nextmv/cloud/application/_run.py

def run_logs_with_polling(
    self: "Application",
    run_id: str,
    verbose: bool = False,
    rich_print: bool = False,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
) -> list[TimestampedRunLog]:
    """
    Get the logs of a 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 sorted by timestamp. 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.

    Returns
    -------
    list[TimestampedRunLog]
        List of timestamped logs of the run.

    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 logs with polling
    >>> logs = app.run_logs_with_polling("run-123", polling_opts)
    >>> for log in logs:
    ...     print(f"[{log.timestamp}] {log.log}")
    [2024-01-01T12:00:00Z] Starting optimization...
    [2024-01-01T12:00:05Z] Found initial solution
    ...
    """

    sleep_duration_hint = 0
    logs = []
    query_params = None

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

        # Perform the actual request to the API.
        response = self.client.request(
            method="GET",
            endpoint=f"{self.endpoint}/runs/{run_id}/logs/live",
            query_params=query_params,
        )
        json_resp = response.json()

        # Get the logs of the current request. Print them if verbose is
        # enabled and append them to the overall logs.
        for resp_log in json_resp.get("items", []):
            log_entry = TimestampedRunLog.from_dict(resp_log)
            if verbose:
                msg = f"[{log_entry.timestamp}] {log_entry.log}"
                if rich_print:
                    rich.print(msg, file=sys.stderr)
                else:
                    log(msg)

            logs.append(log_entry)

        # We are done asking for logs if the run is in a final state.
        status_v2 = StatusV2(json_resp.get("status_v2", "none"))
        if status_v2 in {
            StatusV2.succeeded,
            StatusV2.failed,
            StatusV2.canceled,
        }:
            return logs, True

        # Store the server's hint for the next sleep duration.
        sleep_duration_hint = json_resp.get("next_available_in_seconds", 0)

        # Update the query parameters for the next request.
        since = json_resp.get("next_page_token")
        if since is not None:
            query_params = {"since": since}

        return logs, False

    def sleep_func() -> float:
        return sleep_duration_hint if sleep_duration_hint > 0 else 0

    polling_options.sleep_duration_func = sleep_func
    logs = poll(polling_options=polling_options, polling_func=polling_func)

    return sorted(logs, key=lambda log: log.timestamp)

run_metadata

run_metadata(run_id: str) -> RunInformation

Get the metadata of a run.

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
HTTPError	If the response status code is not 2xx.

Examples:

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

Source code in nextmv/nextmv/cloud/application/_run.py

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

    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
    ------
    requests.HTTPError
        If the response status code is not 2xx.

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

    response = self.client.request(
        method="GET",
        endpoint=f"{self.endpoint}/runs/{run_id}/metadata",
    )

    info = RunInformation.from_dict(response.json())
    info.console_url = self.__console_url(info.id)

    return info

run_result

run_result(
    run_id: str, output_dir_path: str | None = "."
) -> RunResult

Get the result of a run.

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
HTTPError	If the response status code is not 2xx.

Examples:

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

Source code in nextmv/nextmv/cloud/application/_run.py

def run_result(self: "Application", run_id: str, output_dir_path: str | None = ".") -> RunResult:
    """
    Get the result of a run.

    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
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    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 run with polling.

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/cloud/application/_run.py

def run_result_with_polling(
    self: "Application",
    run_id: str,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
    output_dir_path: str | None = ".",
) -> RunResult:
    """
    Get the result of a run with polling.

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

track_run

track_run(
    tracked_run: TrackedRun,
    instance_id: str | None = None,
    configuration: RunConfiguration
    | dict[str, Any]
    | None = None,
) -> str

Track an external run.

This method allows you to register in Nextmv a run that happened elsewhere, as though it were executed in the Nextmv platform. Having information about a run in Nextmv is useful for things like experimenting and testing.

Please read the documentation on the TrackedRun class carefully, as there are important considerations to take into account when using this method. For example, if you intend to upload JSON input/output, use the input/output attributes of the TrackedRun class. On the other hand, if you intend to track files-based input/output, use the input_dir_path/output_dir_path attributes of the TrackedRun class.

PARAMETER	DESCRIPTION
tracked_run	The run to track. TYPE: TrackedRun
instance_id	Optional instance ID if you want to associate your tracked run with an instance. TYPE: Optional[str] DEFAULT: None
configuration	Configuration to use for the run. This can be a cloud.RunConfiguration object or a dict. If the object is used, then the .to_dict() method is applied to extract the configuration. TYPE: RunConfiguration | dict[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
str	The ID of the run that was tracked.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.
ValueError	If the tracked run does not have an input or output.

Examples:

>>> from nextmv.cloud import Application
>>> from nextmv import TrackedRun
>>> app = Application(id="app_123")
>>> tracked_run = TrackedRun(input={"data": [...]}, output={"solution": [...]})
>>> run_id = app.track_run(tracked_run)

Source code in nextmv/nextmv/cloud/application/_run.py

def track_run(
    self: "Application",
    tracked_run: TrackedRun,
    instance_id: str | None = None,
    configuration: RunConfiguration | dict[str, Any] | None = None,
) -> str:
    """
    Track an external run.

    This method allows you to register in Nextmv a run that happened
    elsewhere, as though it were executed in the Nextmv platform. Having
    information about a run in Nextmv is useful for things like
    experimenting and testing.

    Please read the documentation on the `TrackedRun` class carefully, as
    there are important considerations to take into account when using this
    method. For example, if you intend to upload JSON input/output, use the
    `input`/`output` attributes of the `TrackedRun` class. On the other
    hand, if you intend to track files-based input/output, use the
    `input_dir_path`/`output_dir_path` attributes of the `TrackedRun`
    class.

    Parameters
    ----------
    tracked_run : TrackedRun
        The run to track.
    instance_id : Optional[str], default=None
        Optional instance ID if you want to associate your tracked run with
        an instance.
    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.

    Returns
    -------
    str
        The ID of the run that was tracked.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    ValueError
        If the tracked run does not have an input or output.

    Examples
    --------
    >>> from nextmv.cloud import Application
    >>> from nextmv import TrackedRun
    >>> app = Application(id="app_123")
    >>> tracked_run = TrackedRun(input={"data": [...]}, output={"solution": [...]})
    >>> run_id = app.track_run(tracked_run)
    """

    input_upload_id = self.__handle_tracked_run_input(tracked_run)

    # Create the external run result and we add properties to it as required.
    external_result = ExternalRunResult(
        status=tracked_run.status.value,
        execution_duration=tracked_run.duration,
    )

    output_upload_id = self.__handle_tracked_output(tracked_run)
    if output_upload_id is not None:
        external_result.output_upload_id = output_upload_id

    logs_upload_id = self.__handle_tracked_run_logs(tracked_run)
    if logs_upload_id is not None:
        external_result.error_upload_id = logs_upload_id

    if tracked_run.error is not None and tracked_run.error != "":
        external_result.error_message = tracked_run.error

    # Handle the statistics upload if provided.
    stats_upload_id = self.__handle_tracked_run_statistics(tracked_run)
    if stats_upload_id is not None:
        external_result.statistics_upload_id = stats_upload_id

    metrics_upload_id = self.__handle_tracked_run_metrics(tracked_run)
    if metrics_upload_id is not None:
        external_result.metrics_upload_id = metrics_upload_id

    # Handle the assets upload if provided.
    assets_upload_id = self.__handle_tracked_run_assets(tracked_run)
    if assets_upload_id is not None:
        external_result.assets_upload_id = assets_upload_id

    return self.new_run(
        upload_id=input_upload_id,
        external_result=external_result,
        instance_id=instance_id,
        name=tracked_run.name,
        description=tracked_run.description,
        configuration=configuration,
    )

track_run_with_result

track_run_with_result(
    tracked_run: TrackedRun,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
    instance_id: str | None = None,
    output_dir_path: str | None = ".",
    configuration: RunConfiguration
    | dict[str, Any]
    | None = None,
) -> RunResult

Track an external run and poll for the result. This is a convenience method that combines the track_run and run_result_with_polling methods. It applies polling logic to check when the run was successfully registered.

PARAMETER	DESCRIPTION
tracked_run	The run to track. TYPE: TrackedRun
polling_options	Options to use when polling for the run result. TYPE: PollingOptions DEFAULT: DEFAULT_POLLING_OPTIONS
instance_id	Optional instance ID if you want to associate your tracked run with an instance. 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: "."
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

RETURNS	DESCRIPTION
RunResult	Result of the run.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.
ValueError	If the tracked run does not have an input or output.
TimeoutError	If the run does not succeed after the polling strategy is exhausted based on time duration.
RuntimeError	If the run does not succeed after the polling strategy is exhausted based on number of tries.

Source code in nextmv/nextmv/cloud/application/_run.py

def track_run_with_result(
    self: "Application",
    tracked_run: TrackedRun,
    polling_options: PollingOptions = DEFAULT_POLLING_OPTIONS,
    instance_id: str | None = None,
    output_dir_path: str | None = ".",
    configuration: RunConfiguration | dict[str, Any] | None = None,
) -> RunResult:
    """
    Track an external run and poll for the result. This is a convenience
    method that combines the `track_run` and `run_result_with_polling`
    methods. It applies polling logic to check when the run was
    successfully registered.

    Parameters
    ----------
    tracked_run : TrackedRun
        The run to track.
    polling_options : PollingOptions
        Options to use when polling for the run result.
    instance_id: Optional[str]
        Optional instance ID if you want to associate your tracked run with
        an instance.
    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.
    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.

    Returns
    -------
    RunResult
        Result of the run.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    ValueError
        If the tracked run does not have an input or output.
    TimeoutError
        If the run does not succeed after the polling strategy is
        exhausted based on time duration.
    RuntimeError
        If the run does not succeed after the polling strategy is
        exhausted based on number of tries.
    """
    run_id = self.track_run(
        tracked_run=tracked_run,
        instance_id=instance_id,
        configuration=configuration,
    )

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

Secrets Management

_secrets

Application mixin for managing app secrets.

ApplicationSecretsMixin

Mixin class for managing app secrets within an application.

delete_secrets_collection

delete_secrets_collection(
    secrets_collection_id: str,
) -> None

Delete a secrets collection.

PARAMETER	DESCRIPTION
secrets_collection_id	ID of the secrets collection to delete. TYPE: str

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> app.delete_secrets_collection("secrets-123")

Source code in nextmv/nextmv/cloud/application/_secrets.py

def delete_secrets_collection(self: "Application", secrets_collection_id: str) -> None:
    """
    Delete a secrets collection.

    Parameters
    ----------
    secrets_collection_id : str
        ID of the secrets collection to delete.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> app.delete_secrets_collection("secrets-123")
    """

    _ = self.client.request(
        method="DELETE",
        endpoint=f"{self.endpoint}/secrets/{secrets_collection_id}",
    )

list_secrets_collections

list_secrets_collections() -> list[
    SecretsCollectionSummary
]

List all secrets collections.

RETURNS	DESCRIPTION
list[SecretsCollectionSummary]	List of all secrets collections associated with this application.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> collections = app.list_secrets_collections()
>>> for collection in collections:
...     print(collection.name)
'API Keys'
'Database Credentials'

Source code in nextmv/nextmv/cloud/application/_secrets.py

def list_secrets_collections(self: "Application") -> list[SecretsCollectionSummary]:
    """
    List all secrets collections.

    Returns
    -------
    list[SecretsCollectionSummary]
        List of all secrets collections associated with this application.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> collections = app.list_secrets_collections()
    >>> for collection in collections:
    ...     print(collection.name)
    'API Keys'
    'Database Credentials'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.endpoint}/secrets",
    )

    return [SecretsCollectionSummary.from_dict(secrets) for secrets in response.json()["items"]]

new_secrets_collection

new_secrets_collection(
    secrets: list[Secret],
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
) -> SecretsCollectionSummary

Create a new secrets collection.

This method creates a new secrets collection with the provided secrets. A secrets collection is a group of key-value pairs that can be used by your application instances during execution. If no secrets are provided, a ValueError is raised. If the id or name parameters are not provided, they will be generated based on a unique ID.

PARAMETER	DESCRIPTION
secrets	List of secrets to use for the secrets collection. Each secret should be an instance of the Secret class containing a key and value. TYPE: list[Secret]
id	ID of the secrets collection. If not provided, a unique ID will be generated. TYPE: str | None DEFAULT: None
name	Name of the secrets collection. If not provided, the ID will be used. TYPE: str | None DEFAULT: None
description	Description of the secrets collection. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
SecretsCollectionSummary	Summary of the secrets collection including its metadata.

RAISES	DESCRIPTION
ValueError	If no secrets are provided.
HTTPError	If the response status code is not 2xx.

Examples:

>>> # Create a new secrets collection with API keys
>>> from nextmv.cloud import Secret
>>> secrets = [
...     Secret(
...          location="API_KEY",
...          value="your-api-key",
...          secret_type=SecretType.ENV,
...     ),
...     Secret(
...          location="DATABASE_URL",
...          value="your-database-url",
...          secret_type=SecretType.ENV,
...     ),
... ]
>>> collection = app.new_secrets_collection(
...     secrets=secrets,
...     id="api-secrets",
...     name="API Secrets",
...     description="Collection of API secrets for external services"
... )
>>> print(collection.id)
'api-secrets'

Source code in nextmv/nextmv/cloud/application/_secrets.py

def new_secrets_collection(
    self: "Application",
    secrets: list[Secret],
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
) -> SecretsCollectionSummary:
    """
    Create a new secrets collection.

    This method creates a new secrets collection with the provided secrets.
    A secrets collection is a group of key-value pairs that can be used by
    your application instances during execution. If no secrets are
    provided, a ValueError is raised. If the `id` or `name` parameters are
    not provided, they will be generated based on a unique ID.

    Parameters
    ----------
    secrets : list[Secret]
        List of secrets to use for the secrets collection. Each secret
        should be an instance of the Secret class containing a key and
        value.
    id : str | None, default=None
        ID of the secrets collection. If not provided, a unique ID will be
        generated.
    name : str | None, default=None
        Name of the secrets collection. If not provided, the ID will be
        used.
    description : Optional[str], default=None
        Description of the secrets collection.

    Returns
    -------
    SecretsCollectionSummary
        Summary of the secrets collection including its metadata.

    Raises
    ------
    ValueError
        If no secrets are provided.
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> # Create a new secrets collection with API keys
    >>> from nextmv.cloud import Secret
    >>> secrets = [
    ...     Secret(
    ...          location="API_KEY",
    ...          value="your-api-key",
    ...          secret_type=SecretType.ENV,
    ...     ),
    ...     Secret(
    ...          location="DATABASE_URL",
    ...          value="your-database-url",
    ...          secret_type=SecretType.ENV,
    ...     ),
    ... ]
    >>> collection = app.new_secrets_collection(
    ...     secrets=secrets,
    ...     id="api-secrets",
    ...     name="API Secrets",
    ...     description="Collection of API secrets for external services"
    ... )
    >>> print(collection.id)
    'api-secrets'
    """

    if len(secrets) == 0:
        raise ValueError("secrets must be provided")

    if id is None or id == "":
        id = safe_id(prefix="secrets")
    if name is None or name == "":
        name = id

    payload = {
        "id": id,
        "name": name,
        "secrets": [secret.to_dict() for secret in secrets],
    }

    if description is not None:
        payload["description"] = description

    response = self.client.request(
        method="POST",
        endpoint=f"{self.endpoint}/secrets",
        payload=payload,
    )

    return SecretsCollectionSummary.from_dict(response.json())

secrets_collection

secrets_collection(
    secrets_collection_id: str,
) -> SecretsCollection

Get a secrets collection.

This method retrieves a secrets collection by its ID. A secrets collection is a group of key-value pairs that can be used by your application instances during execution.

PARAMETER	DESCRIPTION
secrets_collection_id	ID of the secrets collection to retrieve. TYPE: str

RETURNS	DESCRIPTION
SecretsCollection	The requested secrets collection, including all secret values and metadata.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> # Retrieve a secrets collection
>>> collection = app.secrets_collection("api-secrets")
>>> print(collection.name)
'API Secrets'
>>> print(len(collection.secrets))
2
>>> for secret in collection.secrets:
...     print(secret.location)
'API_KEY'
'DATABASE_URL'

Source code in nextmv/nextmv/cloud/application/_secrets.py

def secrets_collection(self: "Application", secrets_collection_id: str) -> SecretsCollection:
    """
    Get a secrets collection.

    This method retrieves a secrets collection by its ID. A secrets collection
    is a group of key-value pairs that can be used by your application
    instances during execution.

    Parameters
    ----------
    secrets_collection_id : str
        ID of the secrets collection to retrieve.

    Returns
    -------
    SecretsCollection
        The requested secrets collection, including all secret values
        and metadata.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> # Retrieve a secrets collection
    >>> collection = app.secrets_collection("api-secrets")
    >>> print(collection.name)
    'API Secrets'
    >>> print(len(collection.secrets))
    2
    >>> for secret in collection.secrets:
    ...     print(secret.location)
    'API_KEY'
    'DATABASE_URL'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.endpoint}/secrets/{secrets_collection_id}",
    )

    return SecretsCollection.from_dict(response.json())

update_secrets_collection

update_secrets_collection(
    secrets_collection_id: str,
    name: str | None = None,
    description: str | None = None,
    secrets: list[Secret | dict[str, Any]] | None = None,
) -> SecretsCollectionSummary

Update a secrets collection.

This method updates an existing secrets collection with new values for name, description, and secrets. A secrets collection is a group of key-value pairs that can be used by your application instances during execution.

PARAMETER	DESCRIPTION
secrets_collection_id	ID of the secrets collection to update. TYPE: str
name	Optional new name for the secrets collection. TYPE: Optional[str] DEFAULT: None
description	Optional new description for the secrets collection. TYPE: Optional[str] DEFAULT: None
secrets	Optional list of secrets to update. Each secret should be an instance of the Secret class containing a key and value. TYPE: Optional[list[Secret | dict[str, Any]]] DEFAULT: None

RETURNS	DESCRIPTION
SecretsCollectionSummary	Summary of the updated secrets collection including its metadata.

RAISES	DESCRIPTION
ValueError	If no secrets are provided.
HTTPError	If the response status code is not 2xx.

Examples:

>>> # Update an existing secrets collection
>>> from nextmv.cloud import Secret
>>> updated_secrets = [
...     Secret(key="API_KEY", value="new-api-key"),
...     Secret(key="DATABASE_URL", value="new-database-url")
... ]
>>> updated_collection = app.update_secrets_collection(
...     secrets_collection_id="api-secrets",
...     name="Updated API Secrets",
...     description="Updated collection of API secrets",
...     secrets=updated_secrets
... )
>>> print(updated_collection.id)
'api-secrets'

Source code in nextmv/nextmv/cloud/application/_secrets.py

def update_secrets_collection(
    self: "Application",
    secrets_collection_id: str,
    name: str | None = None,
    description: str | None = None,
    secrets: list[Secret | dict[str, Any]] | None = None,
) -> SecretsCollectionSummary:
    """
    Update a secrets collection.

    This method updates an existing secrets collection with new values for name,
    description, and secrets. A secrets collection is a group of key-value pairs
    that can be used by your application instances during execution.

    Parameters
    ----------
    secrets_collection_id : str
        ID of the secrets collection to update.
    name : Optional[str], default=None
        Optional new name for the secrets collection.
    description : Optional[str], default=None
        Optional new description for the secrets collection.
    secrets : Optional[list[Secret | dict[str, Any]]], default=None
        Optional list of secrets to update. Each secret should be an
        instance of the Secret class containing a key and value.

    Returns
    -------
    SecretsCollectionSummary
        Summary of the updated secrets collection including its metadata.

    Raises
    ------
    ValueError
        If no secrets are provided.
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> # Update an existing secrets collection
    >>> from nextmv.cloud import Secret
    >>> updated_secrets = [
    ...     Secret(key="API_KEY", value="new-api-key"),
    ...     Secret(key="DATABASE_URL", value="new-database-url")
    ... ]
    >>> updated_collection = app.update_secrets_collection(
    ...     secrets_collection_id="api-secrets",
    ...     name="Updated API Secrets",
    ...     description="Updated collection of API secrets",
    ...     secrets=updated_secrets
    ... )
    >>> print(updated_collection.id)
    'api-secrets'
    """

    collection = self.secrets_collection(secrets_collection_id)
    collection_dict = collection.to_dict()
    payload = collection_dict.copy()

    if name is not None:
        payload["name"] = name
    if description is not None:
        payload["description"] = description
    if secrets is not None and len(secrets) > 0:
        secrets_dicts = []
        for ix, secret in enumerate(secrets):
            if isinstance(secret, dict):
                secrets_dicts.append(secret)
            elif isinstance(secret, Secret):
                secrets_dicts.append(secret.to_dict())
            else:
                raise ValueError(f"secret at index {ix} must be either a Secret or dict object")

        payload["secrets"] = secrets_dicts

    response = self.client.request(
        method="PUT",
        endpoint=f"{self.endpoint}/secrets/{secrets_collection_id}",
        payload=payload,
    )

    return SecretsCollectionSummary.from_dict(response.json())

Shadow Test Management

_shadow

Application mixin for managing shadow tests.

ApplicationShadowMixin

Mixin class for managing shadow tests within an application.

delete_shadow_test

delete_shadow_test(shadow_test_id: str) -> None

Delete a shadow test.

Deletes a shadow test along with all the associated information, such as its runs.

PARAMETER	DESCRIPTION
shadow_test_id	ID of the shadow test to delete. TYPE: str

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> app.delete_shadow_test("shadow-123")

Source code in nextmv/nextmv/cloud/application/_shadow.py

def delete_shadow_test(self: "Application", shadow_test_id: str) -> None:
    """
    Delete a shadow test.

    Deletes a shadow test along with all the associated information,
    such as its runs.

    Parameters
    ----------
    shadow_test_id : str
        ID of the shadow test to delete.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> app.delete_shadow_test("shadow-123")
    """

    _ = self.client.request(
        method="DELETE",
        endpoint=f"{self.experiments_endpoint}/shadow/{shadow_test_id}",
    )

list_shadow_tests

list_shadow_tests() -> list[ShadowTest]

List all shadow tests.

RETURNS	DESCRIPTION
list[ShadowTest]	List of shadow tests.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_shadow.py

def list_shadow_tests(self: "Application") -> list[ShadowTest]:
    """
    List all shadow tests.

    Returns
    -------
    list[ShadowTest]
        List of shadow tests.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/shadow",
    )

    return [ShadowTest.from_dict(shadow_test) for shadow_test in response.json()]

new_shadow_test

new_shadow_test(
    comparisons: dict[str, list[str]],
    termination_events: TerminationEvents,
    shadow_test_id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    start_events: StartEvents | None = None,
) -> ShadowTest

Create a new shadow test in draft mode. Shadow tests are experiments that run instances in parallel to compare their results.

Use the comparisons parameter to define how to set up instance comparisons. The keys of the comparisons dictionary are the baseline instance IDs, and the values are the candidate lists of instance IDs to compare against the respective baseline.

You may specify start_events to make the shadow test start at a specific time. Alternatively, you may use the start_shadow_test method to start the test.

The termination_events parameter is required and provides control over when the shadow test should terminate. Alternatively, you may use the stop_shadow_test method to stop the test.

PARAMETER	DESCRIPTION
comparisons	Dictionary defining the baseline and candidate instance IDs for comparison. The keys are baseline instance IDs, and the values are lists of candidate instance IDs to compare against the respective baseline. TYPE: dict[str, list[str]]
termination_events	Termination events for the shadow test. TYPE: TerminationEvents
shadow_test_id	ID of the shadow test. Will be generated if not provided. TYPE: Optional[str] DEFAULT: None
name	Name of the shadow test. If not provided, the ID will be used as the name. TYPE: Optional[str] DEFAULT: None
description	Optional description of the shadow test. TYPE: Optional[str] DEFAULT: None
start_events	Start events for the shadow test. TYPE: Optional[StartEvents] DEFAULT: None

RETURNS	DESCRIPTION
ShadowTest	The created shadow test.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_shadow.py

def new_shadow_test(
    self: "Application",
    comparisons: dict[str, list[str]],
    termination_events: TerminationEvents,
    shadow_test_id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    start_events: StartEvents | None = None,
) -> ShadowTest:
    """
    Create a new shadow test in draft mode. Shadow tests are experiments
    that run instances in parallel to compare their results.

    Use the `comparisons` parameter to define how to set up instance
    comparisons. The keys of the `comparisons` dictionary are the baseline
    instance IDs, and the values are the candidate lists of instance IDs to
    compare against the respective baseline.

    You may specify `start_events` to make the shadow test start at a
    specific time. Alternatively, you may use the `start_shadow_test`
    method to start the test.

    The `termination_events` parameter is required and provides control
    over when the shadow test should terminate. Alternatively, you may use
    the `stop_shadow_test` method to stop the test.

    Parameters
    ----------
    comparisons : dict[str, list[str]]
        Dictionary defining the baseline and candidate instance IDs for
        comparison. The keys are baseline instance IDs, and the values are
        lists of candidate instance IDs to compare against the respective
        baseline.
    termination_events : TerminationEvents
        Termination events for the shadow test.
    shadow_test_id : Optional[str]
        ID of the shadow test. Will be generated if not provided.
    name : Optional[str]
        Name of the shadow test. If not provided, the ID will be used as
        the name.
    description : Optional[str]
        Optional description of the shadow test.
    start_events : Optional[StartEvents]
        Start events for the shadow test.

    Returns
    -------
    ShadowTest
        The created shadow test.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    # Generate ID if not provided
    if shadow_test_id is None:
        shadow_test_id = safe_id("shadow")

    # Use ID as name if name not provided
    if name is None or name == "":
        name = shadow_test_id

    payload = {
        "id": shadow_test_id,
        "name": name,
        "comparisons": comparisons,
        "termination_events": termination_events.to_dict(),
    }

    if description is not None:
        payload["description"] = description
    if start_events is not None:
        payload["start_events"] = start_events.to_dict()

    response = self.client.request(
        method="POST",
        endpoint=f"{self.experiments_endpoint}/shadow",
        payload=payload,
    )

    return ShadowTest.from_dict(response.json())

shadow_test

shadow_test(shadow_test_id: str) -> ShadowTest

Get a shadow test. This method also returns the runs of the shadow test under the .runs attribute.

PARAMETER	DESCRIPTION
shadow_test_id	ID of the shadow test. TYPE: str

RETURNS	DESCRIPTION
ShadowTest	The requested shadow test details.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> shadow_test = app.shadow_test("shadow-123")
>>> print(shadow_test.name)
'My Shadow Test'

Source code in nextmv/nextmv/cloud/application/_shadow.py

def shadow_test(self: "Application", shadow_test_id: str) -> ShadowTest:
    """
    Get a shadow test. This method also returns the runs of the shadow
    test under the `.runs` attribute.

    Parameters
    ----------
    shadow_test_id : str
        ID of the shadow test.

    Returns
    -------
    ShadowTest
        The requested shadow test details.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> shadow_test = app.shadow_test("shadow-123")
    >>> print(shadow_test.name)
    'My Shadow Test'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/shadow/{shadow_test_id}",
    )

    exp = ShadowTest.from_dict(response.json())

    runs_response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/shadow/{shadow_test_id}/runs",
    )

    runs = [Run.from_dict(run) for run in runs_response.json().get("runs", [])]
    exp.runs = runs

    return exp

shadow_test_metadata

shadow_test_metadata(
    shadow_test_id: str,
) -> ShadowTestMetadata

Get metadata for a shadow test.

PARAMETER	DESCRIPTION
shadow_test_id	ID of the shadow test. TYPE: str

RETURNS	DESCRIPTION
ShadowTestMetadata	The requested shadow test metadata.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> metadata = app.shadow_test_metadata("shadow-123")
>>> print(metadata.name)
'My Shadow Test'

Source code in nextmv/nextmv/cloud/application/_shadow.py

def shadow_test_metadata(self: "Application", shadow_test_id: str) -> ShadowTestMetadata:
    """
    Get metadata for a shadow test.

    Parameters
    ----------
    shadow_test_id : str
        ID of the shadow test.

    Returns
    -------
    ShadowTestMetadata
        The requested shadow test metadata.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> metadata = app.shadow_test_metadata("shadow-123")
    >>> print(metadata.name)
    'My Shadow Test'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/shadow/{shadow_test_id}/metadata",
    )

    return ShadowTestMetadata.from_dict(response.json())

start_shadow_test

start_shadow_test(shadow_test_id: str) -> None

Start a shadow test. Create a shadow test in draft mode using the new_shadow_test method, then use this method to start the test.

PARAMETER	DESCRIPTION
shadow_test_id	ID of the shadow test to start. TYPE: str

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_shadow.py

def start_shadow_test(self: "Application", shadow_test_id: str) -> None:
    """
    Start a shadow test. Create a shadow test in draft mode using the
    `new_shadow_test` method, then use this method to start the test.

    Parameters
    ----------
    shadow_test_id : str
        ID of the shadow test to start.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    _ = self.client.request(
        method="PUT",
        endpoint=f"{self.experiments_endpoint}/shadow/{shadow_test_id}/start",
    )

stop_shadow_test

stop_shadow_test(
    shadow_test_id: str, intent: StopIntent
) -> None

Stop a shadow test. The test should already have started before using this method.

PARAMETER	DESCRIPTION
shadow_test_id	ID of the shadow test to stop. TYPE: str
intent	Intent for stopping the shadow test. TYPE: StopIntent

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_shadow.py

def stop_shadow_test(self: "Application", shadow_test_id: str, intent: StopIntent) -> None:
    """
    Stop a shadow test. The test should already have started before using
    this method.

    Parameters
    ----------
    shadow_test_id : str
        ID of the shadow test to stop.
    intent : StopIntent
        Intent for stopping the shadow test.
    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    payload = {
        "intent": intent.value,
    }

    _ = self.client.request(
        method="PUT",
        endpoint=f"{self.experiments_endpoint}/shadow/{shadow_test_id}/stop",
        payload=payload,
    )

update_shadow_test

update_shadow_test(
    shadow_test_id: str,
    name: str | None = None,
    description: str | None = None,
) -> ShadowTest

Update a shadow test.

PARAMETER	DESCRIPTION
shadow_test_id	ID of the shadow test to update. TYPE: str
name	Optional name of the shadow test. TYPE: Optional[str] DEFAULT: None
description	Optional description of the shadow test. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
ShadowTest	The information with the updated shadow test.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_shadow.py

def update_shadow_test(
    self: "Application",
    shadow_test_id: str,
    name: str | None = None,
    description: str | None = None,
) -> ShadowTest:
    """
    Update a shadow test.

    Parameters
    ----------
    shadow_test_id : str
        ID of the shadow test to update.
    name : Optional[str], default=None
        Optional name of the shadow test.
    description : Optional[str], default=None
        Optional description of the shadow test.

    Returns
    -------
    ShadowTest
        The information with the updated shadow test.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    payload = {}

    if name is not None:
        payload["name"] = name
    if description is not None:
        payload["description"] = description

    response = self.client.request(
        method="PATCH",
        endpoint=f"{self.experiments_endpoint}/shadow/{shadow_test_id}",
        payload=payload,
    )

    return ShadowTest.from_dict(response.json())

Switchback Test Management

_switchback

Application mixin for managing switchback tests.

ApplicationSwitchbackMixin

Mixin class for managing switchback tests within an application.

delete_switchback_test

delete_switchback_test(switchback_test_id: str) -> None

Delete a switchback test.

Deletes a switchback test along with all the associated information, such as its runs.

PARAMETER	DESCRIPTION
switchback_test_id	ID of the switchback test to delete. TYPE: str

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> app.delete_switchback_test("switchback-123")

Source code in nextmv/nextmv/cloud/application/_switchback.py

def delete_switchback_test(self: "Application", switchback_test_id: str) -> None:
    """
    Delete a switchback test.

    Deletes a switchback test along with all the associated information,
    such as its runs.

    Parameters
    ----------
    switchback_test_id : str
        ID of the switchback test to delete.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> app.delete_switchback_test("switchback-123")
    """

    _ = self.client.request(
        method="DELETE",
        endpoint=f"{self.experiments_endpoint}/switchback/{switchback_test_id}",
    )

list_switchback_tests

list_switchback_tests() -> list[SwitchbackTest]

List all switchback tests.

RETURNS	DESCRIPTION
list[SwitchbackTest]	List of switchback tests.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_switchback.py

def list_switchback_tests(self: "Application") -> list[SwitchbackTest]:
    """
    List all switchback tests.

    Returns
    -------
    list[SwitchbackTest]
        List of switchback tests.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/switchback",
    )

    return [SwitchbackTest.from_dict(switchback_test) for switchback_test in response.json().get("items", [])]

new_switchback_test

new_switchback_test(
    comparison: TestComparisonSingle,
    unit_duration_minutes: float,
    units: int,
    switchback_test_id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    start: datetime | None = None,
) -> SwitchbackTest

Create a new switchback test in draft mode. Switchback tests are experiments that alternate between different instances over specified time intervals.

Use the comparison parameter to define how to set up the instance comparison. The test will alternate between the baseline and candidate instances defined in the comparison.

You may specify start to make the switchback test start at a specific time. Alternatively, you may use the start_switchback_test method to start the test.

PARAMETER	DESCRIPTION
comparison	Comparison defining the baseline and candidate instances. TYPE: TestComparisonSingle
unit_duration_minutes	Duration of each interval in minutes. The value must be between 1 and 10080. TYPE: float
units	Total number of intervals in the switchback test. The value must be between 1 and 1000. TYPE: int
switchback_test_id	Optional ID for the switchback test. Will be generated if not provided. TYPE: Optional[str] DEFAULT: None
name	Optional name of the switchback test. If not provided, the ID will be used as the name. TYPE: Optional[str] DEFAULT: None
description	Optional description of the switchback test. TYPE: Optional[str] DEFAULT: None
start	Optional scheduled start time for the switchback test. TYPE: Optional[datetime] DEFAULT: None

RETURNS	DESCRIPTION
SwitchbackTest	The created switchback test.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_switchback.py

def new_switchback_test(
    self: "Application",
    comparison: TestComparisonSingle,
    unit_duration_minutes: float,
    units: int,
    switchback_test_id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    start: datetime | None = None,
) -> SwitchbackTest:
    """
    Create a new switchback test in draft mode. Switchback tests are
    experiments that alternate between different instances over specified
    time intervals.

    Use the `comparison` parameter to define how to set up the instance
    comparison. The test will alternate between the baseline and candidate
    instances defined in the comparison.

    You may specify `start` to make the switchback test start at a
    specific time. Alternatively, you may use the `start_switchback_test`
    method to start the test.

    Parameters
    ----------
    comparison : TestComparisonSingle
        Comparison defining the baseline and candidate instances.
    unit_duration_minutes : float
        Duration of each interval in minutes. The value must be between 1
        and 10080.
    units : int
        Total number of intervals in the switchback test. The value must be
        between 1 and 1000.
    switchback_test_id : Optional[str], default=None
        Optional ID for the switchback test. Will be generated if not
        provided.
    name : Optional[str], default=None
        Optional name of the switchback test. If not provided, the ID will
        be used as the name.
    description : Optional[str], default=None
        Optional description of the switchback test.
    start : Optional[datetime], default=None
        Optional scheduled start time for the switchback test.

    Returns
    -------
    SwitchbackTest
        The created switchback test.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    if unit_duration_minutes < 1 or unit_duration_minutes > 10080:
        raise ValueError("unit_duration_minutes must be between 1 and 10080")

    if units < 1 or units > 1000:
        raise ValueError("units must be between 1 and 1000")

    # Generate ID if not provided
    if switchback_test_id is None:
        switchback_test_id = safe_id("switchback")

    # Use ID as name if name not provided
    if name is None or name == "":
        name = switchback_test_id

    payload = {
        "id": switchback_test_id,
        "name": name,
        "comparison": comparison.to_dict(),
        "generate_random_plan": {
            "unit_duration_minutes": unit_duration_minutes,
            "units": units,
        },
    }

    if description is not None:
        payload["description"] = description
    if start is not None:
        payload["generate_random_plan"]["start"] = start.isoformat()

    response = self.client.request(
        method="POST",
        endpoint=f"{self.experiments_endpoint}/switchback",
        payload=payload,
    )

    return SwitchbackTest.from_dict(response.json())

start_switchback_test

start_switchback_test(switchback_test_id: str) -> None

Start a switchback test. Create a switchback test in draft mode using the new_switchback_test method, then use this method to start the test.

PARAMETER	DESCRIPTION
switchback_test_id	ID of the switchback test to start. TYPE: str

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_switchback.py

def start_switchback_test(self: "Application", switchback_test_id: str) -> None:
    """
    Start a switchback test. Create a switchback test in draft mode using the
    `new_switchback_test` method, then use this method to start the test.

    Parameters
    ----------
    switchback_test_id : str
        ID of the switchback test to start.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    _ = self.client.request(
        method="PUT",
        endpoint=f"{self.experiments_endpoint}/switchback/{switchback_test_id}/start",
    )

stop_switchback_test

stop_switchback_test(
    switchback_test_id: str, intent: StopIntent
) -> None

Stop a switchback test. The test should already have started before using this method.

PARAMETER	DESCRIPTION
switchback_test_id	ID of the switchback test to stop. TYPE: str
intent	Intent for stopping the switchback test. TYPE: StopIntent

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_switchback.py

def stop_switchback_test(self: "Application", switchback_test_id: str, intent: StopIntent) -> None:
    """
    Stop a switchback test. The test should already have started before using
    this method.

    Parameters
    ----------
    switchback_test_id : str
        ID of the switchback test to stop.

    intent : StopIntent
        Intent for stopping the switchback test.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    payload = {
        "intent": intent.value,
    }

    _ = self.client.request(
        method="PUT",
        endpoint=f"{self.experiments_endpoint}/switchback/{switchback_test_id}/stop",
        payload=payload,
    )

switchback_test

switchback_test(switchback_test_id: str) -> SwitchbackTest

Get a switchback test. This method also returns the runs of the switchback test under the .runs attribute.

PARAMETER	DESCRIPTION
switchback_test_id	ID of the switchback test. TYPE: str

RETURNS	DESCRIPTION
SwitchbackTest	The requested switchback test details.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> switchback_test = app.switchback_test("switchback-123")
>>> print(switchback_test.name)
'My Switchback Test'

Source code in nextmv/nextmv/cloud/application/_switchback.py

def switchback_test(self: "Application", switchback_test_id: str) -> SwitchbackTest:
    """
    Get a switchback test. This method also returns the runs of the switchback
    test under the `.runs` attribute.

    Parameters
    ----------
    switchback_test_id : str
        ID of the switchback test.

    Returns
    -------
    SwitchbackTest
        The requested switchback test details.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> switchback_test = app.switchback_test("switchback-123")
    >>> print(switchback_test.name)
    'My Switchback Test'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/switchback/{switchback_test_id}",
    )

    exp = SwitchbackTest.from_dict(response.json())

    runs_response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/switchback/{switchback_test_id}/runs",
    )

    runs = [Run.from_dict(run) for run in runs_response.json().get("runs", [])]
    exp.runs = runs

    return exp

switchback_test_metadata

switchback_test_metadata(
    switchback_test_id: str,
) -> SwitchbackTestMetadata

Get metadata for a switchback test.

PARAMETER	DESCRIPTION
switchback_test_id	ID of the switchback test. TYPE: str

RETURNS	DESCRIPTION
SwitchbackTestMetadata	The requested switchback test metadata.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> metadata = app.switchback_test_metadata("switchback-123")
>>> print(metadata.name)
'My Switchback Test'

Source code in nextmv/nextmv/cloud/application/_switchback.py

def switchback_test_metadata(self: "Application", switchback_test_id: str) -> SwitchbackTestMetadata:
    """
    Get metadata for a switchback test.

    Parameters
    ----------
    switchback_test_id : str
        ID of the switchback test.

    Returns
    -------
    SwitchbackTestMetadata
        The requested switchback test metadata.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> metadata = app.switchback_test_metadata("switchback-123")
    >>> print(metadata.name)
    'My Switchback Test'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.experiments_endpoint}/switchback/{switchback_test_id}/metadata",
    )

    return SwitchbackTestMetadata.from_dict(response.json())

update_switchback_test

update_switchback_test(
    switchback_test_id: str,
    name: str | None = None,
    description: str | None = None,
) -> SwitchbackTest

Update a switchback test.

PARAMETER	DESCRIPTION
switchback_test_id	ID of the switchback test to update. TYPE: str
name	Optional name of the switchback test. TYPE: Optional[str] DEFAULT: None
description	Optional description of the switchback test. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
SwitchbackTest	The information with the updated switchback test.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Source code in nextmv/nextmv/cloud/application/_switchback.py

def update_switchback_test(
    self: "Application",
    switchback_test_id: str,
    name: str | None = None,
    description: str | None = None,
) -> SwitchbackTest:
    """
    Update a switchback test.

    Parameters
    ----------
    switchback_test_id : str
        ID of the switchback test to update.
    name : Optional[str], default=None
        Optional name of the switchback test.
    description : Optional[str], default=None
        Optional description of the switchback test.

    Returns
    -------
    SwitchbackTest
        The information with the updated switchback test.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    """

    payload = {}

    if name is not None:
        payload["name"] = name
    if description is not None:
        payload["description"] = description

    response = self.client.request(
        method="PATCH",
        endpoint=f"{self.experiments_endpoint}/switchback/{switchback_test_id}",
        payload=payload,
    )

    return SwitchbackTest.from_dict(response.json())

Version Management

_version

Application mixin for managing app versions.

ApplicationVersionMixin

Mixin class for managing app versions within an application.

delete_version

delete_version(version_id: str) -> None

Delete a version.

Permanently removes the specified version from the application.

PARAMETER	DESCRIPTION
version_id	ID of the version to delete. TYPE: str

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> app.delete_version("v1.0.0")  # Permanently deletes the version

Source code in nextmv/nextmv/cloud/application/_version.py

def delete_version(self: "Application", version_id: str) -> None:
    """
    Delete a version.

    Permanently removes the specified version from the application.

    Parameters
    ----------
    version_id : str
        ID of the version to delete.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> app.delete_version("v1.0.0")  # Permanently deletes the version
    """

    _ = self.client.request(
        method="DELETE",
        endpoint=f"{self.endpoint}/versions/{version_id}",
    )

list_versions

list_versions() -> list[Version]

List all versions.

RETURNS	DESCRIPTION
list[Version]	List of all versions associated with this application.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> versions = app.list_versions()
>>> for version in versions:
...     print(version.name)
'v1.0.0'
'v1.1.0'

Source code in nextmv/nextmv/cloud/application/_version.py

def list_versions(self: "Application") -> list[Version]:
    """
    List all versions.

    Returns
    -------
    list[Version]
        List of all versions associated with this application.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> versions = app.list_versions()
    >>> for version in versions:
    ...     print(version.name)
    'v1.0.0'
    'v1.1.0'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.endpoint}/versions",
    )

    return [Version.from_dict(version) for version in response.json()]

new_version

new_version(
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    exist_ok: bool = False,
) -> Version

Create a new version using the latest pushed executable.

This method creates a new version of the application using the current development binary. Application versions represent different iterations of your application's code and configuration that can be deployed.

PARAMETER	DESCRIPTION
id	ID of the version. If not provided, a unique ID will be generated. TYPE: Optional[str] DEFAULT: None
name	Name of the version. If not provided, a name will be generated. TYPE: Optional[str] DEFAULT: None
description	Description of the version. If not provided, a description will be generated. TYPE: Optional[str] DEFAULT: None
exist_ok	If True and a version with the same ID already exists, return the existing version instead of creating a new one. If True, the 'id' parameter must be provided. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
Version	The newly created (or existing) version.

RAISES	DESCRIPTION
ValueError	If exist_ok is True and id is None.
HTTPError	If the response status code is not 2xx.

Examples:

>>> # Create a new version
>>> version = app.new_version(
...     id="v1.0.0",
...     name="Initial Release",
...     description="First stable version"
... )
>>> print(version.id)
'v1.0.0'

>>> # Get or create a version with exist_ok
>>> version = app.new_version(
...     id="v1.0.0",
...     exist_ok=True
... )

Source code in nextmv/nextmv/cloud/application/_version.py

def new_version(
    self: "Application",
    id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    exist_ok: bool = False,
) -> Version:
    """
    Create a new version using the latest pushed executable.

    This method creates a new version of the application using the current development
    binary. Application versions represent different iterations of your application's
    code and configuration that can be deployed.

    Parameters
    ----------
    id : Optional[str], default=None
        ID of the version. If not provided, a unique ID will be generated.
    name : Optional[str], default=None
        Name of the version. If not provided, a name will be generated.
    description : Optional[str], default=None
        Description of the version. If not provided, a description will be generated.
    exist_ok : bool, default=False
        If True and a version with the same ID already exists,
        return the existing version instead of creating a new one.
        If True, the 'id' parameter must be provided.

    Returns
    -------
    Version
        The newly created (or existing) version.

    Raises
    ------
    ValueError
        If exist_ok is True and id is None.
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> # Create a new version
    >>> version = app.new_version(
    ...     id="v1.0.0",
    ...     name="Initial Release",
    ...     description="First stable version"
    ... )
    >>> print(version.id)
    'v1.0.0'

    >>> # Get or create a version with exist_ok
    >>> version = app.new_version(
    ...     id="v1.0.0",
    ...     exist_ok=True
    ... )
    """

    if exist_ok and (id is None or id == ""):
        raise ValueError("If exist_ok is True, id must be provided")

    if exist_ok and self.version_exists(version_id=id):
        return self.version(version_id=id)

    if id is None or id == "":
        id = safe_id(prefix="version")

    if name is None or name == "":
        name = id

    payload = {
        "id": id,
        "name": name,
    }

    if description is not None:
        payload["description"] = description

    response = self.client.request(
        method="POST",
        endpoint=f"{self.endpoint}/versions",
        payload=payload,
    )

    return Version.from_dict(response.json())

update_version

update_version(
    version_id: str,
    name: str | None = None,
    description: str | None = None,
) -> Version

Update a version.

This method updates a specific version of the application. It mimics a PATCH operation by allowing you to update only the name and/or description fields while preserving all other fields.

PARAMETER	DESCRIPTION
version_id	ID of the version to update. TYPE: str
name	Optional new name for the version. TYPE: Optional[str] DEFAULT: None
description	Optional new description for the version. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
Version	The updated version object.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> # Update a version's name
>>> updated = app.update_version("v1.0.0", name="Version 1.0")
>>> print(updated.name)
'Version 1.0'

>>> # Update a version's description
>>> updated = app.update_version("v1.0.0", description="Initial release")
>>> print(updated.description)
'Initial release'

Source code in nextmv/nextmv/cloud/application/_version.py

def update_version(
    self: "Application",
    version_id: str,
    name: str | None = None,
    description: str | None = None,
) -> Version:
    """
    Update a version.

    This method updates a specific version of the application. It mimics a
    PATCH operation by allowing you to update only the name and/or description
    fields while preserving all other fields.

    Parameters
    ----------
    version_id : str
        ID of the version to update.
    name : Optional[str], default=None
        Optional new name for the version.
    description : Optional[str], default=None
        Optional new description for the version.

    Returns
    -------
    Version
        The updated version object.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> # Update a version's name
    >>> updated = app.update_version("v1.0.0", name="Version 1.0")
    >>> print(updated.name)
    'Version 1.0'

    >>> # Update a version's description
    >>> updated = app.update_version("v1.0.0", description="Initial release")
    >>> print(updated.description)
    'Initial release'
    """

    version = self.version(version_id=version_id)
    version_dict = version.to_dict()
    payload = version_dict.copy()

    if name is not None:
        payload["name"] = name
    if description is not None:
        payload["description"] = description

    response = self.client.request(
        method="PUT",
        endpoint=f"{self.endpoint}/versions/{version_id}",
        payload=payload,
    )

    return Version.from_dict(response.json())

version

version(version_id: str) -> Version

Get a version.

Retrieves a specific version of the application by its ID. Application versions represent different iterations of your application's code and configuration.

PARAMETER	DESCRIPTION
version_id	ID of the version to retrieve. TYPE: str

RETURNS	DESCRIPTION
Version	The version object containing details about the requested application version.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> # Retrieve a specific version
>>> version = app.version("v1.0.0")
>>> print(version.id)
'v1.0.0'
>>> print(version.name)
'Initial Release'

Source code in nextmv/nextmv/cloud/application/_version.py

def version(self: "Application", version_id: str) -> Version:
    """
    Get a version.

    Retrieves a specific version of the application by its ID. Application versions
    represent different iterations of your application's code and configuration.

    Parameters
    ----------
    version_id : str
        ID of the version to retrieve.

    Returns
    -------
    Version
        The version object containing details about the requested application version.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> # Retrieve a specific version
    >>> version = app.version("v1.0.0")
    >>> print(version.id)
    'v1.0.0'
    >>> print(version.name)
    'Initial Release'
    """

    response = self.client.request(
        method="GET",
        endpoint=f"{self.endpoint}/versions/{version_id}",
    )

    return Version.from_dict(response.json())

version_exists

version_exists(version_id: str) -> bool

Check if a version exists.

This method checks if a specific version of the application exists by attempting to retrieve it. It handles HTTP errors for non-existent versions and returns a boolean indicating existence.

PARAMETER	DESCRIPTION
version_id	ID of the version to check for existence. TYPE: str

RETURNS	DESCRIPTION
bool	True if the version exists, False otherwise.

RAISES	DESCRIPTION
HTTPError	If an HTTP error occurs that is not related to the non-existence of the version.

Examples:

>>> # Check if a version exists
>>> exists = app.version_exists("v1.0.0")
>>> if exists:
...     print("Version exists!")
... else:
...     print("Version does not exist.")

Source code in nextmv/nextmv/cloud/application/_version.py

def version_exists(self: "Application", version_id: str) -> bool:
    """
    Check if a version exists.

    This method checks if a specific version of the application exists by
    attempting to retrieve it. It handles HTTP errors for non-existent versions
    and returns a boolean indicating existence.

    Parameters
    ----------
    version_id : str
        ID of the version to check for existence.

    Returns
    -------
    bool
        True if the version exists, False otherwise.

    Raises
    ------
    requests.HTTPError
        If an HTTP error occurs that is not related to the non-existence
        of the version.

    Examples
    --------
    >>> # Check if a version exists
    >>> exists = app.version_exists("v1.0.0")
    >>> if exists:
    ...     print("Version exists!")
    ... else:
    ...     print("Version does not exist.")
    """

    try:
        self.version(version_id=version_id)
        return True
    except requests.HTTPError as e:
        if _is_not_exist_error(e):
            return False
        raise e