Community module¶

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

def clone_community_app(
    client: Client,
    app: str,
    directory: str | None = None,
    version: str | None = LATEST_VERSION,
    verbose: bool = False,
    rich_print: bool = False,
    should_register: 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.
    should_register : bool, optional
        Whether to register the cloned app in the local registry. Default is False.
    """
    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 verbose:
        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,
            )
        else:
            log(
                f"✅ Successfully cloned the {app} community app, using version {original_version} "
                f"in path: {full_destination}."
            )

    _register_cloned_app(
        app=app,
        full_destination=full_destination,
        verbose=verbose,
        rich_print=rich_print,
        should_register=should_register,
    )