Community Module

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

community

This module contains functionality for working with Nextmv community apps.

Community apps are pre-built decision models. They are maintained in the following GitHub repository: https://github.com/nextmv-io/community-apps

CLASS	DESCRIPTION
CommunityApp	Representation of a Nextmv Cloud Community App.

FUNCTION	DESCRIPTION
list_community_apps	List the available Nextmv community apps.
clone_community_app	Clone a community app locally.

CommunityApp

Bases: BaseModel

Information about a Nextmv community app.

You can import the CommunityApp class directly from cloud:

from nextmv.cloud import CommunityApp

PARAMETER	DESCRIPTION
app_versions	Available versions of the community app. TYPE: list[str]
description	Description of the community app. TYPE: str
latest_app_version	The latest version of the community app. TYPE: str
latest_marketplace_version	The latest version of the community app in the Nextmv Marketplace. TYPE: str
marketplace_versions	Available versions of the community app in the Nextmv Marketplace. TYPE: list[str]
name	Name of the community app. TYPE: str
app_type	Type of the community app. TYPE: str

app_type class-attribute instance-attribute

app_type: str = Field(
    serialization_alias="type",
    validation_alias=AliasChoices("type", "app_type"),
)

Type of the community app.

app_versions class-attribute instance-attribute

app_versions: list[str] | None = None

Available versions of the community app.

description instance-attribute

description: str

Description of the community app.

has_version

has_version(version: str) -> bool

Check if the community app has the specified version.

PARAMETER	DESCRIPTION
version	The version to check. TYPE: str

RETURNS	DESCRIPTION
bool	True if the app has the specified version, False otherwise.

Source code in nextmv/nextmv/cloud/community.py

def has_version(self, version: str) -> bool:
    """
    Check if the community app has the specified version.

    Parameters
    ----------
    version : str
        The version to check.

    Returns
    -------
    bool
        True if the app has the specified version, False otherwise.
    """

    if version == LATEST_VERSION:
        version = self.latest_app_version

    if self.app_versions is not None and version in self.app_versions:
        return True

    return False

latest_app_version class-attribute instance-attribute

latest_app_version: str | None = None

The latest version of the community app.

latest_marketplace_version class-attribute instance-attribute

latest_marketplace_version: str | None = None

The latest version of the community app in the Nextmv Marketplace.

marketplace_versions class-attribute instance-attribute

marketplace_versions: list[str] | None = None

Available versions of the community app in the Nextmv Marketplace.

name instance-attribute

name: str

Name of the community app.

LATEST_VERSION module-attribute

LATEST_VERSION = 'latest'

clone_community_app

clone_community_app(
    client: Client,
    app: str,
    directory: str | None = None,
    version: str | None = LATEST_VERSION,
    verbose: bool = False,
    rich_print: bool = False,
) -> None

Clone a community app locally.

By default, the latest version will be used. You can specify a version with the version parameter, and customize the output directory with the directory parameter. If you want to list the available apps, use the list_community_apps function.

You can import the clone_community_app function directly from cloud:

from nextmv.cloud import clone_community_app

PARAMETER	DESCRIPTION
client	The Nextmv Cloud client to use for the request. TYPE: Client
app	The name of the community app to clone. TYPE: str
directory	The directory in which to clone the app. Default is the name of the app at current directory. TYPE: str | None DEFAULT: None
version	The version of the community app to clone. Default is latest. TYPE: str | None DEFAULT: LATEST_VERSION
verbose	Whether to print verbose output. TYPE: bool DEFAULT: False
rich_print	Whether to use rich printing for output messages. TYPE: bool DEFAULT: False

Source code in nextmv/nextmv/cloud/community.py

def clone_community_app(
    client: Client,
    app: str,
    directory: str | None = None,
    version: str | None = LATEST_VERSION,
    verbose: bool = False,
    rich_print: bool = False,
) -> None:
    """
    Clone a community app locally.

    By default, the `latest` version will be used. You can
    specify a version with the `version` parameter, and customize the output
    directory with the `directory` parameter. If you want to list the available
    apps, use the `list_community_apps` function.

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

    ```python
    from nextmv.cloud import clone_community_app
    ```

    Parameters
    ----------
    client : Client
        The Nextmv Cloud client to use for the request.
    app : str
        The name of the community app to clone.
    directory : str | None, optional
        The directory in which to clone the app. Default is the name of the app at current directory.
    version : str | None, optional
        The version of the community app to clone. Default is `latest`.
    verbose : bool, optional
        Whether to print verbose output.
    rich_print : bool, optional
        Whether to use rich printing for output messages.
    """
    comm_app = _find_app(client, app)

    if version is not None and version == "":
        raise ValueError("`version` cannot be an empty string.")

    if not comm_app.has_version(version):
        raise ValueError(f"Community app '{app}' does not have version '{version}'.")

    original_version = version
    if version == LATEST_VERSION:
        version = comm_app.latest_app_version

    # Clean and normalize directory path in an OS-independent way
    if directory is not None and directory != "":
        destination = os.path.normpath(directory)
    else:
        destination = app

    full_destination = _get_valid_path(destination, os.stat)
    os.makedirs(full_destination, exist_ok=True)

    tarball = f"{app}_{version}.tar.gz"
    s3_file_path = f"{app}/{version}/{tarball}"
    downloaded_object = _download_object(
        client=client,
        file=s3_file_path,
        path="community-apps",
        output_dir=full_destination,
        output_file=tarball,
    )

    # Extract the tarball to a temporary directory to handle nested structure
    with tempfile.TemporaryDirectory() as temp_dir:
        with tarfile.open(downloaded_object, "r:gz") as tar:
            tar.extractall(path=temp_dir)

        # Find the extracted directory (typically the app name)
        extracted_items = os.listdir(temp_dir)
        if len(extracted_items) == 1 and os.path.isdir(os.path.join(temp_dir, extracted_items[0])):
            # Move contents from the extracted directory to full_destination
            extracted_dir = os.path.join(temp_dir, extracted_items[0])
            for item in os.listdir(extracted_dir):
                shutil.move(os.path.join(extracted_dir, item), full_destination)
        else:
            # If structure is unexpected, move everything directly
            for item in extracted_items:
                shutil.move(os.path.join(temp_dir, item), full_destination)

    # Remove the tarball after extraction
    os.remove(downloaded_object)

    if not verbose:
        return

    if rich_print:
        rich.print(
            f":white_check_mark: Successfully cloned the [magenta]{app}[/magenta] community app, "
            f"using version [magenta]{original_version}[/magenta] in path: [magenta]{full_destination}[/magenta].",
            file=sys.stderr,
        )
        return

    log(
        f"✅ Successfully cloned the {app} community app, using version {original_version} in path: {full_destination}."
    )

list_community_apps

list_community_apps(client: Client) -> list[CommunityApp]

List the available Nextmv community apps.

You can import the list_community_apps function directly from cloud:

from nextmv.cloud import list_community_apps

PARAMETER	DESCRIPTION
client	The Nextmv Cloud client to use for the request. TYPE: Client

RETURNS	DESCRIPTION
list[CommunityApp]	A list of available community apps.

Source code in nextmv/nextmv/cloud/community.py

def list_community_apps(client: Client) -> list[CommunityApp]:
    """
    List the available Nextmv community apps.

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

    ```python
    from nextmv.cloud import list_community_apps
    ```

    Parameters
    ----------
    client : Client
        The Nextmv Cloud client to use for the request.

    Returns
    -------
    list[CommunityApp]
        A list of available community apps.
    """

    manifest = _download_manifest(client)
    dict_apps = manifest.get("apps", [])
    apps = [CommunityApp.from_dict(app) for app in dict_apps]

    return apps