Package module¶

def package(
    app_dir: str,
    manifest: Manifest,
    model: Model | None = None,
    model_configuration: ModelConfiguration | None = None,
    verbose: bool = False,
    rich_print: bool = False,
    no_cache: bool = False,
) -> tuple[str, str]:
    """
    Package the app into a tarball.

    Parameters
    ----------
    app_dir : str
        The directory of the application to package.
    manifest : Manifest
        The app manifest describing the application type, files, and dependencies.
    model : Model, optional
        The Python model to encode and include in the package.
    model_configuration : ModelConfiguration, optional
        The configuration for encoding the Python model.
    verbose : bool, optional
        Whether to print verbose logs.
    rich_print : bool, optional
        Whether to use rich printing for verbose logs.
    no_cache : bool, default=False
        When working with Python, dependencies are cached to speed up
        subsequent pushes. Setting no_cache to True will skip using the
        cache and force a fresh build of all dependencies. This is useful
        when you want to ensure that you are pushing the most up-to-date
        versions of your dependencies, or if you are encountering issues
        with the cache and want to rule it out as a potential cause.

    Returns
    -------
    tuple of (str, str)
        A tuple containing the path to the resulting ``app.tar.gz`` file and
        the path to the output directory that holds it.  The caller is
        responsible for cleaning up the output directory when it is no longer
        needed.
    """

    with tempfile.TemporaryDirectory(prefix="nextmv-temp-") as temp_dir:
        deps_tar: Path | None = None
        output_dir: str | None = None
        success = False
        try:
            if manifest.type == ManifestType.PYTHON:
                deps_tar = _handle_python(app_dir, manifest, model, model_configuration, verbose, rich_print, no_cache)

            found, missing, files = find_files(app_dir, manifest.files)
            manifest.confirm_mandatory_files(present_files=found)

            if len(missing) > 0:
                raise Exception(f"could not find files listed in manifest: {', '.join(missing)}")

            manifest.to_yaml(temp_dir)
            _copy_manifest_files(files, temp_dir, verbose, rich_print)

            if manifest.type == ManifestType.PYTHON:
                _cleanup_python_model(app_dir, model_configuration, verbose)

            output_dir = tempfile.mkdtemp(prefix="nextmv-build-out-")
            tar_file, _ = _compress_and_report(deps_tar, temp_dir, output_dir, verbose, rich_print)

            success = True
            return tar_file, output_dir
        finally:
            if deps_tar is not None:
                shutil.rmtree(str(deps_tar.parent), ignore_errors=True)
            if not success and output_dir is not None:
                shutil.rmtree(output_dir, ignore_errors=True)

def run_build_command(
    app_dir: str,
    manifest_build: ManifestBuild | None = None,
    verbose: bool = False,
    rich_print: bool = False,
) -> None:
    """
    Run the build command specified in the manifest.

    Parameters
    ----------
    app_dir : str
        The directory of the application, used as the working directory when
        running the build command.
    manifest_build : ManifestBuild, optional
        The build configuration from the manifest.  If ``None`` or if
        ``manifest_build.command`` is empty, this function is a no-op.
    verbose : bool, optional
        Whether to print verbose logs.
    rich_print : bool, optional
        Whether to use rich printing for verbose logs.

    Raises
    ------
    Exception
        If the build command exits with a non-zero return code.
    """

    if manifest_build is None or manifest_build.command is None or manifest_build.command == "":
        return

    elements = manifest_build.command.split(" ")
    command_str = " ".join(elements)

    if verbose:
        if rich_print:
            rich.print(f":construction: Running build command: [magenta]{command_str}[/magenta]", file=sys.stderr)
        else:
            log(f'🚧 Running build command: "{command_str}"')
    try:
        result = subprocess.run(
            elements,
            env={**os.environ, **manifest_build.environment_to_dict()},
            check=True,
            text=True,
            capture_output=True,
            cwd=app_dir,
        )

    except subprocess.CalledProcessError as e:
        raise Exception(f"error running build command: {e.stderr}") from e

    if verbose and result.stdout.strip():
        log(result.stdout.rstrip("\n"))

def run_pre_push_command(
    app_dir: str,
    pre_push_command: str | None = None,
    verbose: bool = False,
    rich_print: bool = False,
) -> None:
    """
    Run the pre-push command specified in the manifest.

    Parameters
    ----------
    app_dir : str
        The directory of the application, used as the working directory when
        running the pre-push command.
    pre_push_command : str, optional
        The shell command to execute before pushing.  If ``None`` or empty,
        this function is a no-op.
    verbose : bool, optional
        Whether to print verbose logs.
    rich_print : bool, optional
        Whether to use rich printing for verbose logs.

    Raises
    ------
    Exception
        If the pre-push command exits with a non-zero return code.
    """

    if pre_push_command is None or pre_push_command == "":
        return

    elements = _get_shell_command_elements(pre_push_command)

    command_str = " ".join(elements)
    if verbose:
        if rich_print:
            rich.print(f":hammer: Running pre-push command: [magenta]{command_str}[/magenta]", file=sys.stderr)
        else:
            log(f'🔨 Running pre-push command: "{command_str}"')
    try:
        result = subprocess.run(
            elements,
            env=os.environ,
            check=True,
            text=True,
            capture_output=True,
            cwd=app_dir,
        )

    except subprocess.CalledProcessError as e:
        raise Exception(f"error running pre-push command: {e.stderr}") from e

    if verbose and result.stdout.strip():
        log(result.stdout.rstrip("\n"))