Stack Recipes

zenml.cli.stack_recipes

Functionality to handle downloading ZenML stacks via the CLI.

GitStackRecipesHandler

Class for the GitStackRecipesHandler that interfaces with the CLI.

Source code in zenml/cli/stack_recipes.py

class GitStackRecipesHandler(object):
    """Class for the `GitStackRecipesHandler` that interfaces with the CLI."""

    def __init__(self) -> None:
        """Create a new GitStackRecipesHandler instance."""
        self.repo_dir = io_utils.get_global_config_directory()
        self.stack_recipes_dir = Path(
            os.path.join(self.repo_dir, STACK_RECIPES_REPO_DIR)
        )
        self.stack_recipe_repo = StackRecipeRepo(self.stack_recipes_dir)

    @property
    def stack_recipes(self) -> List[StackRecipe]:
        """Property that contains a list of stack recipes.

        Returns:
            A list of stack recipes.
        """
        return [
            StackRecipe(name, Path(os.path.join(self.stack_recipes_dir, name)))
            for name in sorted(os.listdir(self.stack_recipes_dir))
            if (
                not name.startswith(".")
                and not name.startswith("__")
                and not name == "LICENSE"
                and not name.endswith(".md")
                and not name.endswith(".sh")
            )
        ]

    def is_stack_recipe(self, stack_recipe_name: Optional[str] = None) -> bool:
        """Checks if the given stack_recipe_name corresponds to a stack_recipe.

        Args:
            stack_recipe_name: The name of the stack_recipe to check.

        Returns:
            Whether the supplied stack_recipe_name corresponds to a
            stack_recipe.
        """
        stack_recipe_dict = {
            recipe.name: recipe for recipe in self.stack_recipes
        }
        if stack_recipe_name:
            if stack_recipe_name in stack_recipe_dict.keys():
                return True

        return False

    def get_stack_recipes(
        self, stack_recipe_name: Optional[str] = None
    ) -> List[StackRecipe]:
        """Method that allows you to get a stack recipe by name.

        If no stack recipe is supplied,  all stack recipes are returned.

        Args:
            stack_recipe_name: Name of an stack recipe.

        Returns:
            A list of stack recipes.

        Raises:
            KeyError: If the supplied stack_recipe_name is not found.
        """
        stack_recipe_dict = {
            recipe.name: recipe
            for recipe in self.stack_recipes
            if recipe.name not in EXCLUDED_RECIPE_DIRS
        }
        if stack_recipe_name:
            if stack_recipe_name in stack_recipe_dict.keys():
                return [stack_recipe_dict[stack_recipe_name]]
            else:
                raise KeyError(
                    f"Stack recipe {stack_recipe_name} does not exist! "
                    f"Available Stack Recipes: {list(stack_recipe_dict)}"
                )
        else:
            return self.stack_recipes

    def pull(
        self,
        branch: str,
        force: bool = False,
    ) -> None:
        """Pulls the stack recipes from the main git stack recipes repository.

        Args:
            branch: The name of the branch to pull from.
            force: Whether to force the pull.
        """
        from git.exc import GitCommandError

        if not self.stack_recipe_repo.is_cloned:
            self.stack_recipe_repo.clone()
        elif force:
            self.stack_recipe_repo.delete()
            self.stack_recipe_repo.clone()

        try:
            self.stack_recipe_repo.checkout(branch=branch)
        except GitCommandError:
            cli_utils.warning(
                f"The specified branch {branch} not found in "
                "repo, falling back to the latest release."
            )
            self.stack_recipe_repo.checkout_latest_release()

    def pull_latest_stack_recipes(self) -> None:
        """Pulls the latest stack recipes from the stack recipes repository."""
        self.pull(
            branch=self.stack_recipe_repo.latest_release_branch, force=True
        )

    @staticmethod
    def copy_stack_recipe(
        stack_recipe: StackRecipe, destination_dir: str
    ) -> None:
        """Copies a stack recipe to the destination_dir.

        Args:
            stack_recipe: The stack recipe to copy.
            destination_dir: The destination directory to copy the recipe to.
        """
        io_utils.create_dir_if_not_exists(destination_dir)
        io_utils.copy_dir(
            str(stack_recipe.path_in_repo), destination_dir, overwrite=True
        )

    @staticmethod
    def clean_current_stack_recipes() -> None:
        """Deletes the stack recipes directory from your working directory."""
        stack_recipes_directory = os.path.join(
            os.getcwd(), "zenml_stack_recipes"
        )
        shutil.rmtree(stack_recipes_directory)

    def get_active_version(self) -> Optional[str]:
        """Returns the active version of the mlops-stacks repository.

        Returns:
            The active version of the repository.
        """
        self.stack_recipe_repo.checkout_latest_release()
        return self.stack_recipe_repo.active_version

stack_recipes: List[zenml.cli.stack_recipes.StackRecipe] property readonly

Property that contains a list of stack recipes.

Returns:

Type	Description
List[zenml.cli.stack_recipes.StackRecipe]	A list of stack recipes.

__init__(self) special

Create a new GitStackRecipesHandler instance.

Source code in zenml/cli/stack_recipes.py

def __init__(self) -> None:
    """Create a new GitStackRecipesHandler instance."""
    self.repo_dir = io_utils.get_global_config_directory()
    self.stack_recipes_dir = Path(
        os.path.join(self.repo_dir, STACK_RECIPES_REPO_DIR)
    )
    self.stack_recipe_repo = StackRecipeRepo(self.stack_recipes_dir)

clean_current_stack_recipes() staticmethod

Deletes the stack recipes directory from your working directory.

Source code in zenml/cli/stack_recipes.py

@staticmethod
def clean_current_stack_recipes() -> None:
    """Deletes the stack recipes directory from your working directory."""
    stack_recipes_directory = os.path.join(
        os.getcwd(), "zenml_stack_recipes"
    )
    shutil.rmtree(stack_recipes_directory)

copy_stack_recipe(stack_recipe, destination_dir) staticmethod

Copies a stack recipe to the destination_dir.

Parameters:

Name	Type	Description	Default
stack_recipe	StackRecipe	The stack recipe to copy.	required
destination_dir	str	The destination directory to copy the recipe to.	required

Source code in zenml/cli/stack_recipes.py

@staticmethod
def copy_stack_recipe(
    stack_recipe: StackRecipe, destination_dir: str
) -> None:
    """Copies a stack recipe to the destination_dir.

    Args:
        stack_recipe: The stack recipe to copy.
        destination_dir: The destination directory to copy the recipe to.
    """
    io_utils.create_dir_if_not_exists(destination_dir)
    io_utils.copy_dir(
        str(stack_recipe.path_in_repo), destination_dir, overwrite=True
    )

get_active_version(self)

Returns the active version of the mlops-stacks repository.

Returns:

Type	Description
Optional[str]	The active version of the repository.

Source code in zenml/cli/stack_recipes.py

def get_active_version(self) -> Optional[str]:
    """Returns the active version of the mlops-stacks repository.

    Returns:
        The active version of the repository.
    """
    self.stack_recipe_repo.checkout_latest_release()
    return self.stack_recipe_repo.active_version

get_stack_recipes(self, stack_recipe_name=None)

Method that allows you to get a stack recipe by name.

If no stack recipe is supplied, all stack recipes are returned.

Parameters:

Name	Type	Description	Default
stack_recipe_name	Optional[str]	Name of an stack recipe.	None

Returns:

Type	Description
List[zenml.cli.stack_recipes.StackRecipe]	A list of stack recipes.

Exceptions:

Type	Description
KeyError	If the supplied stack_recipe_name is not found.

Source code in zenml/cli/stack_recipes.py

def get_stack_recipes(
    self, stack_recipe_name: Optional[str] = None
) -> List[StackRecipe]:
    """Method that allows you to get a stack recipe by name.

    If no stack recipe is supplied,  all stack recipes are returned.

    Args:
        stack_recipe_name: Name of an stack recipe.

    Returns:
        A list of stack recipes.

    Raises:
        KeyError: If the supplied stack_recipe_name is not found.
    """
    stack_recipe_dict = {
        recipe.name: recipe
        for recipe in self.stack_recipes
        if recipe.name not in EXCLUDED_RECIPE_DIRS
    }
    if stack_recipe_name:
        if stack_recipe_name in stack_recipe_dict.keys():
            return [stack_recipe_dict[stack_recipe_name]]
        else:
            raise KeyError(
                f"Stack recipe {stack_recipe_name} does not exist! "
                f"Available Stack Recipes: {list(stack_recipe_dict)}"
            )
    else:
        return self.stack_recipes

is_stack_recipe(self, stack_recipe_name=None)

Checks if the given stack_recipe_name corresponds to a stack_recipe.

Parameters:

Name	Type	Description	Default
stack_recipe_name	Optional[str]	The name of the stack_recipe to check.	None

Returns:

Type	Description
bool	Whether the supplied stack_recipe_name corresponds to a stack_recipe.

Source code in zenml/cli/stack_recipes.py

def is_stack_recipe(self, stack_recipe_name: Optional[str] = None) -> bool:
    """Checks if the given stack_recipe_name corresponds to a stack_recipe.

    Args:
        stack_recipe_name: The name of the stack_recipe to check.

    Returns:
        Whether the supplied stack_recipe_name corresponds to a
        stack_recipe.
    """
    stack_recipe_dict = {
        recipe.name: recipe for recipe in self.stack_recipes
    }
    if stack_recipe_name:
        if stack_recipe_name in stack_recipe_dict.keys():
            return True

    return False

pull(self, branch, force=False)

Pulls the stack recipes from the main git stack recipes repository.

Parameters:

Name	Type	Description	Default
branch	str	The name of the branch to pull from.	required
force	bool	Whether to force the pull.	False

Source code in zenml/cli/stack_recipes.py

def pull(
    self,
    branch: str,
    force: bool = False,
) -> None:
    """Pulls the stack recipes from the main git stack recipes repository.

    Args:
        branch: The name of the branch to pull from.
        force: Whether to force the pull.
    """
    from git.exc import GitCommandError

    if not self.stack_recipe_repo.is_cloned:
        self.stack_recipe_repo.clone()
    elif force:
        self.stack_recipe_repo.delete()
        self.stack_recipe_repo.clone()

    try:
        self.stack_recipe_repo.checkout(branch=branch)
    except GitCommandError:
        cli_utils.warning(
            f"The specified branch {branch} not found in "
            "repo, falling back to the latest release."
        )
        self.stack_recipe_repo.checkout_latest_release()

pull_latest_stack_recipes(self)

Pulls the latest stack recipes from the stack recipes repository.

Source code in zenml/cli/stack_recipes.py

def pull_latest_stack_recipes(self) -> None:
    """Pulls the latest stack recipes from the stack recipes repository."""
    self.pull(
        branch=self.stack_recipe_repo.latest_release_branch, force=True
    )

LocalStackRecipe

Class to encapsulate the local recipe that can be run from the CLI.

Source code in zenml/cli/stack_recipes.py

class LocalStackRecipe:
    """Class to encapsulate the local recipe that can be run from the CLI."""

    def __init__(self, path: Path, name: str) -> None:
        """Create a new LocalStack instance.

        Args:
            name: The name of the stack, specifically the name of the folder
                  on git
            path: Path at which the stack is installed
        """
        self.name = name
        self.path = path

    def is_present(self) -> bool:
        """Checks if the stack_recipe exists at the given path.

        Returns:
            True if the stack_recipe exists at the given path, else False.
        """
        return fileio.isdir(str(self.path))

    @property
    def locals_content(self) -> str:
        """Returns the locals.tf content associated with a particular recipe.

        Returns:
            The locals.tf content associated with a particular recipe.

        Raises:
            ValueError: If the locals.tf file is not found.
            FileNotFoundError: If the locals.tf file is not one of the options.
        """
        locals_file = os.path.join(self.path, "locals.tf")
        try:
            with open(locals_file) as locals:
                locals_content = locals.read()
            return locals_content
        except FileNotFoundError:
            if fileio.exists(str(self.path)) and fileio.isdir(str(self.path)):
                raise ValueError(f"No locals.tf file found in " f"{self.path}")
            else:
                raise FileNotFoundError(
                    f"Recipe {self.name} is not one of the available options."
                    f"\n"
                    f"To list all available recipes, type: `zenml stack recipe "
                    f"list`"
                )

locals_content: str property readonly

Returns the locals.tf content associated with a particular recipe.

Returns:

Type	Description
str	The locals.tf content associated with a particular recipe.

Exceptions:

Type	Description
ValueError	If the locals.tf file is not found.
FileNotFoundError	If the locals.tf file is not one of the options.

__init__(self, path, name) special

Create a new LocalStack instance.

Parameters:

Name	Type	Description	Default
name	str	The name of the stack, specifically the name of the folder on git	required
path	Path	Path at which the stack is installed	required

Source code in zenml/cli/stack_recipes.py

def __init__(self, path: Path, name: str) -> None:
    """Create a new LocalStack instance.

    Args:
        name: The name of the stack, specifically the name of the folder
              on git
        path: Path at which the stack is installed
    """
    self.name = name
    self.path = path

is_present(self)

Checks if the stack_recipe exists at the given path.

Returns:

Type	Description
bool	True if the stack_recipe exists at the given path, else False.

Source code in zenml/cli/stack_recipes.py

def is_present(self) -> bool:
    """Checks if the stack_recipe exists at the given path.

    Returns:
        True if the stack_recipe exists at the given path, else False.
    """
    return fileio.isdir(str(self.path))

StackRecipe

Class for all stack recipe objects.

Source code in zenml/cli/stack_recipes.py

class StackRecipe:
    """Class for all stack recipe objects."""

    def __init__(self, name: str, path_in_repo: Path) -> None:
        """Create a new StackRecipe instance.

        Args:
            name: The name of the recipe, specifically the name of the folder
                  on git
            path_in_repo: Path to the local recipe within the global zenml
                  folder.
        """
        self.name = name
        self.path_in_repo = path_in_repo

    @property
    def readme_content(self) -> str:
        """Returns the README content associated with a particular recipe.

        Returns:
            The README content associated with a particular recipe.

        Raises:
            ValueError: If the README file is not found.
            FileNotFoundError: If the README file is not one of the options.
        """
        readme_file = os.path.join(self.path_in_repo, "README.md")
        try:
            with open(readme_file) as readme:
                readme_content = readme.read()
            return readme_content
        except FileNotFoundError:
            if fileio.exists(str(self.path_in_repo)) and fileio.isdir(
                str(self.path_in_repo)
            ):
                raise ValueError(
                    f"No README.md file found in " f"{self.path_in_repo}"
                )
            else:
                raise FileNotFoundError(
                    f"Recipe {self.name} is not one of the available options."
                    f"\n"
                    f"To list all available recipes, type: `zenml stack recipe "
                    f"list`"
                )

readme_content: str property readonly

Returns the README content associated with a particular recipe.

Returns:

Type	Description
str	The README content associated with a particular recipe.

Exceptions:

Type	Description
ValueError	If the README file is not found.
FileNotFoundError	If the README file is not one of the options.

__init__(self, name, path_in_repo) special

Create a new StackRecipe instance.

Parameters:

Name	Type	Description	Default
name	str	The name of the recipe, specifically the name of the folder on git	required
path_in_repo	Path	Path to the local recipe within the global zenml folder.	required

Source code in zenml/cli/stack_recipes.py

def __init__(self, name: str, path_in_repo: Path) -> None:
    """Create a new StackRecipe instance.

    Args:
        name: The name of the recipe, specifically the name of the folder
              on git
        path_in_repo: Path to the local recipe within the global zenml
              folder.
    """
    self.name = name
    self.path_in_repo = path_in_repo

StackRecipeRepo

Class that represents the stack recipes repo.

Source code in zenml/cli/stack_recipes.py

class StackRecipeRepo:
    """Class that represents the stack recipes repo."""

    def __init__(self, cloning_path: Path) -> None:
        """Create a new StackRecipeRepo instance.

        Args:
            cloning_path: Path to the local stack recipe repository.

        Raises:
            GitNotFoundError: If git is not installed.
        """
        self.cloning_path = cloning_path

        try:
            from git.exc import InvalidGitRepositoryError, NoSuchPathError
            from git.repo.base import Repo
        except ImportError as e:
            logger.error(
                "In order to use the CLI tool to interact with our recipes, "
                "you need to have an installation of Git on your machine."
            )
            raise GitNotFoundError(e)

        try:
            self.repo = Repo(self.cloning_path)
        except NoSuchPathError or InvalidGitRepositoryError:
            self.repo = None  # type: ignore
            logger.debug(
                f"`Cloning_path`: {self.cloning_path} was empty, "
                "Automatically cloning the recipes."
            )
            self.clone()
            self.checkout_latest_release()

    @property
    def active_version(self) -> Optional[str]:
        """Returns the active version of the repository.

        In case a release branch is checked out, this property returns
        that version as a string, else `None` is returned.

        Returns:
            The active version of the repository.
        """
        for branch in self.repo.heads:
            branch_name = cast(str, branch.name)
            if (
                branch_name.startswith("release/")
                and branch.commit == self.repo.head.commit
            ):
                return branch_name[len("release/") :]

        return None

    @property
    def latest_release_branch(self) -> str:
        """Returns the name of the latest release branch.

        Returns:
            The name of the latest release branch.
        """
        from packaging.version import Version, parse

        tags = sorted(
            self.repo.tags,
            key=lambda t: t.commit.committed_datetime,  # type: ignore
        )

        if not tags:
            return "main"

        latest_tag = parse(tags[-1].name)
        if type(latest_tag) is not Version:
            return "main"

        latest_release_version: str = tags[-1].name
        return f"release/{latest_release_version}"

    @property
    def is_cloned(self) -> bool:
        """Returns whether we have already cloned the repository.

        Returns:
            Whether we have already cloned the repository.
        """
        return self.cloning_path.exists()

    def clone(self) -> None:
        """Clones repo to `cloning_path`.

        If you break off the operation with a `KeyBoardInterrupt` before the
        cloning is completed, this method will delete whatever was partially
        downloaded from your system.
        """
        self.cloning_path.mkdir(parents=True, exist_ok=False)
        try:
            from git.repo.base import Repo

            logger.info(f"Downloading recipes to {self.cloning_path}")
            self.repo = Repo.clone_from(
                STACK_RECIPES_GITHUB_REPO, self.cloning_path, branch="main"
            )
        except KeyboardInterrupt:
            self.delete()
            logger.error("Canceled download of recipes.. Rolled back.")

    def delete(self) -> None:
        """Delete `cloning_path` if it exists.

        Raises:
            AssertionError: If `cloning_path` does not exist.
        """
        if self.cloning_path.exists():
            shutil.rmtree(self.cloning_path)
        else:
            raise AssertionError(
                f"Cannot delete the stack recipes repository from "
                f"{self.cloning_path} as it does not exist."
            )

    def checkout(self, branch: str) -> None:
        """Checks out a specific branch or tag of the repository.

        Args:
            branch: The name of the branch or tag to check out.
        """
        logger.info(f"Checking out branch: {branch}")
        self.repo.git.checkout(branch)

    def checkout_latest_release(self) -> None:
        """Checks out the latest release of the repository."""
        self.checkout(branch=self.latest_release_branch)

active_version: Optional[str] property readonly

Returns the active version of the repository.

In case a release branch is checked out, this property returns that version as a string, else None is returned.

Returns:

Type	Description
Optional[str]	The active version of the repository.

is_cloned: bool property readonly

Returns whether we have already cloned the repository.

Returns:

Type	Description
bool	Whether we have already cloned the repository.

latest_release_branch: str property readonly

Returns the name of the latest release branch.

Returns:

Type	Description
str	The name of the latest release branch.

__init__(self, cloning_path) special

Create a new StackRecipeRepo instance.

Parameters:

Name	Type	Description	Default
cloning_path	Path	Path to the local stack recipe repository.	required

Exceptions:

Type	Description
GitNotFoundError	If git is not installed.

Source code in zenml/cli/stack_recipes.py

def __init__(self, cloning_path: Path) -> None:
    """Create a new StackRecipeRepo instance.

    Args:
        cloning_path: Path to the local stack recipe repository.

    Raises:
        GitNotFoundError: If git is not installed.
    """
    self.cloning_path = cloning_path

    try:
        from git.exc import InvalidGitRepositoryError, NoSuchPathError
        from git.repo.base import Repo
    except ImportError as e:
        logger.error(
            "In order to use the CLI tool to interact with our recipes, "
            "you need to have an installation of Git on your machine."
        )
        raise GitNotFoundError(e)

    try:
        self.repo = Repo(self.cloning_path)
    except NoSuchPathError or InvalidGitRepositoryError:
        self.repo = None  # type: ignore
        logger.debug(
            f"`Cloning_path`: {self.cloning_path} was empty, "
            "Automatically cloning the recipes."
        )
        self.clone()
        self.checkout_latest_release()

checkout(self, branch)

Checks out a specific branch or tag of the repository.

Parameters:

Name	Type	Description	Default
branch	str	The name of the branch or tag to check out.	required

Source code in zenml/cli/stack_recipes.py

def checkout(self, branch: str) -> None:
    """Checks out a specific branch or tag of the repository.

    Args:
        branch: The name of the branch or tag to check out.
    """
    logger.info(f"Checking out branch: {branch}")
    self.repo.git.checkout(branch)

checkout_latest_release(self)

Checks out the latest release of the repository.

Source code in zenml/cli/stack_recipes.py

def checkout_latest_release(self) -> None:
    """Checks out the latest release of the repository."""
    self.checkout(branch=self.latest_release_branch)

clone(self)

Clones repo to cloning_path.

If you break off the operation with a KeyBoardInterrupt before the cloning is completed, this method will delete whatever was partially downloaded from your system.

Source code in zenml/cli/stack_recipes.py

def clone(self) -> None:
    """Clones repo to `cloning_path`.

    If you break off the operation with a `KeyBoardInterrupt` before the
    cloning is completed, this method will delete whatever was partially
    downloaded from your system.
    """
    self.cloning_path.mkdir(parents=True, exist_ok=False)
    try:
        from git.repo.base import Repo

        logger.info(f"Downloading recipes to {self.cloning_path}")
        self.repo = Repo.clone_from(
            STACK_RECIPES_GITHUB_REPO, self.cloning_path, branch="main"
        )
    except KeyboardInterrupt:
        self.delete()
        logger.error("Canceled download of recipes.. Rolled back.")

delete(self)

Delete cloning_path if it exists.

Exceptions:

Type	Description
AssertionError	If cloning_path does not exist.

Source code in zenml/cli/stack_recipes.py

def delete(self) -> None:
    """Delete `cloning_path` if it exists.

    Raises:
        AssertionError: If `cloning_path` does not exist.
    """
    if self.cloning_path.exists():
        shutil.rmtree(self.cloning_path)
    else:
        raise AssertionError(
            f"Cannot delete the stack recipes repository from "
            f"{self.cloning_path} as it does not exist."
        )