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

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.

PARAMETER	DESCRIPTION
`client` ¶	Client to use for interacting with the Nextmv Cloud API. TYPE: `Client`
`id` ¶	ID of the application. TYPE: `str`
`default_instance_id` ¶	Default instance ID to use for submitting runs. TYPE: `str` DEFAULT: `None`
`endpoint` ¶	Base endpoint for the application. TYPE: `str` DEFAULT: `"v1/applications/{id}"`
`experiments_endpoint` ¶	Base endpoint for the experiments in the application. TYPE: `str` DEFAULT: `"{base}/experiments"`

Examples:

>>> from nextmv.cloud import Client, Application
>>> client = Client(api_key="your-api-key")
>>> app = Application(client=client, id="your-app-id")
>>> # Retrieve app information
>>> instances = app.list_instances()

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,
    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. TYPE: `str`
`id` ¶	ID of the application. Will be generated if not provided. TYPE: `str` DEFAULT: `None`
`description` ¶	Description of the application. TYPE: `str` DEFAULT: `None`
`is_workflow` ¶	Whether the application is a Decision Workflow. TYPE: `bool` 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.

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,
    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
        Name of the application.
    id : str, optional
        ID of the application. Will be generated if not provided.
    description : str, optional
        Description of the application.
    is_workflow : bool, optional
        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.

    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 id is None:
        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())

    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,
    no_version: bool = False,
    version_id: str | None = None,
    version_name: str | None = None,
    version_description: str | None = None,
) -> 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.

The default behavior of this function is to create a new application version after the app has been pushed. You can set the no_version argument to True to skip this step. The version_id, version_name, and version_description arguments can be used to customize the version that is created. If the version_id is not specified, a randomly generated ID will be used. If the version_name is not specified, a generic name with a timestamp will be used. Lastly, if no description is specified, then a generic description will also be used.

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`
`no_version` ¶	If True, do not create a new version after pushing the app. TYPE: `bool` DEFAULT: `False`
`version_id` ¶	ID of the version to create after pushing the app. If None, a unique ID will be generated. TYPE: `Optional[str]` DEFAULT: `None`
`version_name` ¶	Name of the version to create after pushing the app. If None, a name with a timestamp will be generated. TYPE: `Optional[str]` DEFAULT: `None`
`version_description` ¶	Description of the version to create after pushing the app. If None, a generic description with a timestamp will be generated. TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`None`

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:

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.

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,
    no_version: bool = False,
    version_id: str | None = None,
    version_name: str | None = None,
    version_description: str | None = None,
) -> 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.

    The default behavior of this function is to create a new application
    version _after_ the app has been pushed. You can set the `no_version`
    argument to `True` to skip this step. The `version_id`, `version_name`,
    and `version_description` arguments can be used to customize the version
    that is created. If the `version_id` is not specified, a randomly
    generated ID will be used. If the `version_name` is not specified, a
    generic name with a timestamp will be used. Lastly, if no description is
    specified, then a generic description will also be used.

    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.
    no_version : bool, default=False
        If True, do not create a new version after pushing the app.
    version_id : Optional[str], default=None
        ID of the version to create after pushing the app. If None, a unique
        ID will be generated.
    version_name : Optional[str], default=None
        Name of the version to create after pushing the app. If None, a name
        with a timestamp will be generated.
    version_description : Optional[str], default=None
        Description of the version to create after pushing the app. If None, a
        generic description with a timestamp will be generated.

    Returns
    -------
    None

    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

    if no_version:
        if verbose:
            if rich_print:
                rich.print(
                    f":white_check_mark: Push completed for Nextmv application [magenta]{self.id}[/magenta] "
                    "without creating a new version.",
                    file=sys.stderr,
                )
            else:
                log("✅ Push completed without creating a new version for Nextmv application.")

        return

    now = datetime.now(timezone.utc)
    if version_id is None:
        version_id = safe_id(prefix="version") + f"-{now.strftime('%Y%m%d-%H%M%S')}"
    if version_name is None:
        version_name = f"Version {version_id}"
    if version_description is None:
        version_description = f"Version created automatically from push at {now.strftime('%Y-%m-%dT%H:%M:%SZ')}"

    version = self.new_version(
        id=version_id,
        name=version_name,
        description=version_description,
    )
    version_dict = version.to_dict()

    if verbose:
        if rich_print:
            rich.print(
                f":white_check_mark: Automatically created new version [magenta]{version.id}[/magenta].",
                file=sys.stderr,
            )
            rich.print_json(data=version_dict)
        else:
            log(f'✅ Automatically created new version "{version.id}".')
            log(json.dumps(version_dict, indent=2))

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

    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`

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,
    id: str,
    metrics: list[Metric | dict[str, Any]],
    name: str,
    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. TYPE: `str`
`metrics` ¶	List of metrics to use for the acceptance test. TYPE: `list[Union[Metric, dict[str, Any]]]`
`name` ¶	Name of the acceptance test. TYPE: `str`
`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(
    self: "Application",
    candidate_instance_id: str,
    baseline_instance_id: str,
    id: str,
    metrics: list[Metric | dict[str, Any]],
    name: str,
    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
        ID of the acceptance test.
    metrics : list[Union[Metric, dict[str, Any]]]
        List of metrics to use for the acceptance test.
    name : str
        Name of the acceptance test.
    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.
    """

    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,
    }
    if description is not None:
        payload["description"] = description
    if id is not None:
        payload["id"] = id

    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,
    id: str,
    metrics: list[Metric | dict[str, Any]],
    name: str,
    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. TYPE: `str`
`metrics` ¶	List of metrics to use for the acceptance test. TYPE: `list[Union[Metric, dict[str, Any]]]`
`name` ¶	Name of the acceptance test. TYPE: `str`
`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,
    id: str,
    metrics: list[Metric | dict[str, Any]],
    name: str,
    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
        ID of the acceptance test.
    metrics : list[Union[Metric, dict[str, Any]]]
        List of metrics to use for the acceptance test.
    name : str
        Name of the acceptance test.
    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,
    )

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,
    input_set_id: str | None = None,
    instance_ids: list[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. TYPE: `str`
`input_set_id` ¶	ID of the input set to use for the batch experiment. TYPE: `str \| None` DEFAULT: `None`
`instance_ids` ¶	List of instance IDs to use for the batch experiment. This argument is deprecated, use `runs` instead. TYPE: `list[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(
    self: "Application",
    name: str,
    input_set_id: str | None = None,
    instance_ids: list[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: str
        Name of the batch experiment.
    input_set_id: str
        ID of the input set to use for the batch experiment.
    instance_ids: list[str]
        List of instance IDs to use for the batch experiment.
        This argument is deprecated, use `runs` instead.
    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.
    """

    payload = {
        "name": name,
    }
    if input_set_id is not None:
        payload["input_set_id"] = input_set_id
    if instance_ids is not None:
        input_set = self.input_set(input_set_id)
        runs = to_runs(instance_ids, input_set)
        payload_runs = [run.to_dict() for run in runs]
        payload["runs"] = payload_runs
    if description is not None:
        payload["description"] = description
    if id is not None:
        payload["id"] = id
    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,
    input_set_id: str | None = None,
    instance_ids: list[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. TYPE: `str`
`input_set_id` ¶	ID of the input set to use for the batch experiment. TYPE: `str \| None` DEFAULT: `None`
`instance_ids` ¶	List of instance IDs to use for the batch experiment. This argument is deprecated, use `runs` instead. TYPE: `list[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,
    input_set_id: str | None = None,
    instance_ids: list[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: str
        Name of the batch experiment.
    input_set_id: str
        ID of the input set to use for the batch experiment.
    instance_ids: list[str]
        List of instance IDs to use for the batch experiment. This argument
        is deprecated, use `runs` instead.
    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,
        instance_ids=instance_ids,
        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(
    id: str,
    name: str,
    scenarios: list[Scenario],
    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
`id` ¶	ID of the scenario test. TYPE: `str`
`name` ¶	Name of the scenario test. TYPE: `str`
`scenarios` ¶	List of scenarios to use for the scenario test. At least one scenario should be provided. TYPE: `list[Scenario]`
`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",
    id: str,
    name: str,
    scenarios: list[Scenario],
    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
    ----------
    id: str
        ID of the scenario test.
    name: str
        Name of the scenario test.
    scenarios: list[Scenario]
        List of scenarios to use for the scenario test. At least one
        scenario should be provided.
    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")

    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(
    id: str,
    name: str,
    scenarios: list[Scenario],
    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
`id` ¶	ID of the scenario test. TYPE: `str`
`name` ¶	Name of the scenario test. TYPE: `str`
`scenarios` ¶	List of scenarios to use for the scenario test. At least one scenario should be provided. TYPE: `list[Scenario]`
`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
`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",
    id: str,
    name: str,
    scenarios: list[Scenario],
    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
    ----------
    id: str
        ID of the scenario test.
    name: str
        Name of the scenario test.
    scenarios: list[Scenario]
        List of scenarios to use for the scenario test. At least one
        scenario should be provided.
    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
    -------
    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(
        id=id,
        name=name,
        scenarios=scenarios,
        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_defintion ¶

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

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:
    """
    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.
    """

    if name is None:
        name = id
    if description is None:
        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())

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.

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,
    name: str,
    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:

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.
Using run_ids. The input set will be created using the list of runs specified by the user.
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 TYPE: `str`
`name` ¶	Name of the input set. TYPE: `str`
`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,
    name: str,
    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
        ID of the input set
    name: str
        Name of the input set.
    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[ExperimentInput]]
        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.
    """

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

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:
        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:
        id = safe_id(prefix="instance")
    if name is None:
        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,
) -> 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`

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

    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

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

    return Instance.from_dict(response.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:

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.
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:
        id = safe_id(prefix="managed-input")
    if name is None:
        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

    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() -> list[Run]

List all runs.

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") -> list[Run]:
    """
    List all runs.

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

    return [Run.from_dict(run) for run in response.json().get("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_format`is not`JSON`. If the final`options`are not of type`dict[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(  # noqa: C901
    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)
    """

    # Get the URL to upload the input to.
    url_input = self.upload_url()

    # Handle the case where the input is being uploaded as files. We need
    # to tar them.
    input_tar_file = ""
    input_dir_path = tracked_run.input_dir_path
    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.")

        input_tar_file = self._package_inputs(input_dir_path)

    # Handle the case where the input is uploaded as Input or a dict.
    upload_input = tracked_run.input
    if upload_input is not None and isinstance(tracked_run.input, Input):
        upload_input = tracked_run.input.data

    # Actually uploads de input.
    self.upload_data(data=upload_input, upload_url=url_input, tar_file=input_tar_file)

    # Get the URL to upload the output to.
    url_output = self.upload_url()

    # Handle the case where the output is being uploaded as files. We need
    # to tar them.
    output_tar_file = ""
    output_dir_path = tracked_run.output_dir_path
    if output_dir_path is not None and output_dir_path != "":
        if not os.path.exists(output_dir_path):
            raise ValueError(f"Directory {output_dir_path} does not exist.")

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

        output_tar_file = self._package_inputs(output_dir_path)

    # Handle the case where the output is uploaded as Output or a dict.
    upload_output = tracked_run.output
    if upload_output is not None and isinstance(tracked_run.output, Output):
        upload_output = tracked_run.output.to_dict()

    # Actually uploads the output.
    self.upload_data(data=upload_output, upload_url=url_output, tar_file=output_tar_file)

    # Create the external run result and appends logs if required.
    external_result = ExternalRunResult(
        output_upload_id=url_output.upload_id,
        status=tracked_run.status.value,
        execution_duration=tracked_run.duration,
    )

    # Handle the stderr logs if provided.
    if tracked_run.logs is not None:
        url_stderr = self.upload_url()
        self.upload_data(data=tracked_run.logs_text(), upload_url=url_stderr)
        external_result.error_upload_id = url_stderr.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 = tracked_run.statistics
    if stats is not None:
        if isinstance(stats, Statistics):
            stats_dict = stats.to_dict()
            stats_dict = {STATISTICS_KEY: stats_dict}
        elif isinstance(stats, dict):
            stats_dict = stats
            if STATISTICS_KEY not in stats_dict:
                stats_dict = {STATISTICS_KEY: stats_dict}
        else:
            raise ValueError("tracked_run.statistics must be either a `Statistics` or `dict` object")

        url_stats = self.upload_url()
        self.upload_data(data=stats_dict, upload_url=url_stats)
        external_result.statistics_upload_id = url_stats.upload_id

    # Handle the assets upload if provided.
    assets = tracked_run.assets
    if assets is not None:
        if isinstance(assets, list):
            assets_list = []
            for ix, asset in enumerate(assets):
                if isinstance(asset, Asset):
                    assets_list.append(asset.to_dict())
                elif isinstance(asset, dict):
                    assets_list.append(asset)
                else:
                    raise ValueError(f"tracked_run.assets, index {ix} must be an `Asset` or `dict` object")
            assets_dict = {ASSETS_KEY: assets_list}
        elif isinstance(assets, dict):
            assets_dict = assets
            if ASSETS_KEY not in assets_dict:
                assets_dict = {ASSETS_KEY: assets_dict}
        else:
            raise ValueError("tracked_run.assets must be either a `list[Asset]`, `list[dict]`, or `dict` object")

        url_assets = self.upload_url()
        self.upload_data(data=assets_dict, upload_url=url_assets)
        external_result.assets_upload_id = url_assets.upload_id

    return self.new_run(
        upload_id=url_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

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

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:
        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:
        id = safe_id(prefix="version")
    if name is None:
        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

    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

Application Module¶

Main Class¶

application ¶

Application ¶

client ¶

id ¶

default_instance_id ¶

endpoint ¶

experiments_endpoint ¶

client class-attribute instance-attribute ¶

created_at class-attribute instance-attribute ¶

default_experiment_instance class-attribute instance-attribute ¶

default_instance_id class-attribute instance-attribute ¶

delete ¶

description class-attribute instance-attribute ¶

endpoint class-attribute instance-attribute ¶

ensembles_endpoint class-attribute instance-attribute ¶

exists staticmethod ¶

client ¶

id ¶

experiments_endpoint class-attribute instance-attribute ¶

get classmethod ¶

client ¶

id ¶

id instance-attribute ¶

locked class-attribute instance-attribute ¶

model_post_init ¶

name class-attribute instance-attribute ¶

new classmethod ¶

client ¶

name ¶

id ¶

description ¶

is_workflow ¶

exist_ok ¶

default_instance_id ¶

default_experiment_instance ¶

push ¶

manifest ¶

app_dir ¶

verbose ¶

model ¶

model_configuration ¶

rich_print ¶

no_version ¶

version_id ¶

version_name ¶

version_description ¶

subscription_id class-attribute instance-attribute ¶

type class-attribute instance-attribute ¶

update ¶

name ¶

description ¶

default_instance_id ¶

default_experiment_instance ¶

updated_at class-attribute instance-attribute ¶

upload_data ¶

upload_url ¶

data ¶

json_configurations ¶

tar_file ¶

upload_large_input ¶

input ¶

upload_url ¶

json_configurations ¶

tar_file ¶

upload_url ¶

ApplicationType ¶

CUSTOM class-attribute instance-attribute ¶

PIPELINE class-attribute instance-attribute ¶

SUBSCRIPTION class-attribute instance-attribute ¶

list_applications ¶

client ¶

Mixin Classes¶

Acceptance Testing¶

_acceptance ¶

ApplicationAcceptanceMixin ¶

acceptance_test ¶

acceptance_test_id ¶

acceptance_test_with_polling ¶

`client` ¶

`id` ¶

`default_instance_id` ¶

`endpoint` ¶

`experiments_endpoint` ¶

client `class-attribute` `instance-attribute` ¶

created_at `class-attribute` `instance-attribute` ¶

default_experiment_instance `class-attribute` `instance-attribute` ¶

default_instance_id `class-attribute` `instance-attribute` ¶

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

endpoint `class-attribute` `instance-attribute` ¶

ensembles_endpoint `class-attribute` `instance-attribute` ¶

exists `staticmethod` ¶

`client` ¶

`id` ¶

experiments_endpoint `class-attribute` `instance-attribute` ¶

get `classmethod` ¶

`client` ¶

`id` ¶

id `instance-attribute` ¶

locked `class-attribute` `instance-attribute` ¶

name `class-attribute` `instance-attribute` ¶

new `classmethod` ¶

`client` ¶

`name` ¶

`id` ¶

`description` ¶

`is_workflow` ¶

`exist_ok` ¶

`default_instance_id` ¶

`default_experiment_instance` ¶

`manifest` ¶

`app_dir` ¶

`verbose` ¶

`model` ¶

`model_configuration` ¶

`rich_print` ¶

`no_version` ¶

`version_id` ¶

`version_name` ¶

`version_description` ¶

subscription_id `class-attribute` `instance-attribute` ¶

type `class-attribute` `instance-attribute` ¶

`name` ¶

`description` ¶

`default_instance_id` ¶

`default_experiment_instance` ¶

updated_at `class-attribute` `instance-attribute` ¶

`upload_url` ¶

`data` ¶

`json_configurations` ¶

`tar_file` ¶

`input` ¶

`upload_url` ¶

`json_configurations` ¶

`tar_file` ¶

CUSTOM `class-attribute` `instance-attribute` ¶

PIPELINE `class-attribute` `instance-attribute` ¶

SUBSCRIPTION `class-attribute` `instance-attribute` ¶

`client` ¶

`acceptance_test_id` ¶

`acceptance_test_id` ¶

`acceptance_test_id` ¶

`candidate_instance_id` ¶

`baseline_instance_id` ¶

`id` ¶

`metrics` ¶

`name` ¶

`input_set_id` ¶

`description` ¶

`candidate_instance_id` ¶

`baseline_instance_id` ¶

`id` ¶

`metrics` ¶

`name` ¶

`input_set_id` ¶

`description` ¶

`polling_options` ¶

`batch_id` ¶

`batch_id` ¶