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 supplied 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 an 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 an 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
        )

    def copy_stack_recipe(
        self, stack_recipe: StackRecipe, destination_dir: str
    ) -> None:
        """Copies an 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
        )

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

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

Deletes the ZenML stack recipes directory from your current working directory.

Source code in zenml/cli/stack_recipes.py

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

copy_stack_recipe(self, stack_recipe, destination_dir)

Copies an 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

def copy_stack_recipe(
    self, stack_recipe: StackRecipe, destination_dir: str
) -> None:
    """Copies an 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_stack_recipes(self, stack_recipe_name=None)

Method that allows you to get an 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 an 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 supplied 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 an 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 supplied 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 an 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 stack that can be run from the CLI.

Source code in zenml/cli/stack_recipes.py

class LocalStackRecipe:
    """Class to encapsulate the local stack 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.
        """
        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."
        )

Terraform

Class to represent terraform applications.

Source code in zenml/cli/stack_recipes.py

class Terraform:
    """Class to represent terraform applications."""

    def __init__(self, path: str) -> None:
        """Creates a client that can be used to call terraform commands.

        Args:
            path: the path to the stack recipe.
        """
        self.tf = python_terraform.Terraform(working_dir=str(path))

    def check_installation(self) -> None:
        """Checks if terraform is installed on the host system.

        Raises:
            RuntimeError: if terraform is not installed.
        """
        if not self._is_installed():
            raise RuntimeError(
                "Terraform is required for stack recipes to run and was not "
                "found installed on your machine. Please visit "
                "https://learn.hashicorp.com/tutorials/terraform/install-cli "
                "to install it."
            )

    def _is_installed(self) -> bool:
        """Checks if terraform is installed on the host system.

        Returns:
            True if terraform is installed, false otherwise.
        """
        # check terraform version to verify installation.
        ret_code, _, _ = self.tf.cmd("-version")
        return bool(ret_code == 0)

    def apply(self) -> str:
        """Function to call terraform init and terraform apply.

        The init call is not repeated if any successful execution has
        happened already, to save time.

        Returns:
            The path to the stack YAML configuration file as a string.

        Raises:
            RuntimeError: if terraform init fails.
        """
        # this directory gets created after a successful init
        previous_run_dir = os.path.join(self.tf.working_dir, ".ignoreme")
        if fileio.exists(previous_run_dir):
            logger.info(
                "Terraform already initialized, "
                "terraform init will not be executed."
            )
        else:
            ret_code, _, _ = self.tf.init(capture_output=False)
            if ret_code != 0:
                raise RuntimeError("The command 'terraform init' failed.")
            fileio.mkdir(previous_run_dir)

        # get variables from the recipe as a python dictionary
        vars = self._get_vars(self.tf.working_dir)

        # once init is successful, call terraform apply
        self.tf.apply(
            var=vars,
            input=False,
            capture_output=False,
            raise_on_error=True,
        )

        # return the path of the stack yaml file
        _, stack_file_path, _ = self.tf.output("stack-yaml-path")
        return str(stack_file_path)

    def _get_vars(self, path: str) -> Any:
        """Get variables as a dictionary from values.tfvars.json.

        Args:
            path: the path to the stack recipe.

        Returns:
            A dictionary of variables to use for the stack recipes
            derived from the tfvars.json file.

        Raises:
            FileNotFoundError: if the values.tfvars.json file is not
                found in the stack recipe.
        """
        import json

        variables_file_path = os.path.join(path, VARIABLES_FILE)
        if not fileio.exists(variables_file_path):
            raise FileNotFoundError(
                "The file values.tfvars.json was not found in the "
                f"recipe's directory at {variables_file_path}. Please "
                "verify if it exists."
            )

        # read values into a dict and return
        with fileio.open(variables_file_path, "r") as f:
            variables = json.load(f)
        return variables

    def set_log_level(self, log_level: str) -> None:
        """Set TF_LOG env var to the log_level provided by the user.

        Args:
            log_level: One of TRACE, DEBUG, INFO, WARN or ERROR to set
                as the log level for terraform CLI.
        """
        os.environ["TF_LOG"] = log_level

    def destroy(self) -> None:
        """Function to call terraform destroy on the given path."""
        self.tf.destroy(
            capture_output=False,
            raise_on_error=True,
            force=python_terraform.IsNotFlagged,
        )

__init__(self, path) special

Creates a client that can be used to call terraform commands.

Parameters:

Name	Type	Description	Default
path	str	the path to the stack recipe.	required

Source code in zenml/cli/stack_recipes.py

def __init__(self, path: str) -> None:
    """Creates a client that can be used to call terraform commands.

    Args:
        path: the path to the stack recipe.
    """
    self.tf = python_terraform.Terraform(working_dir=str(path))

apply(self)

Function to call terraform init and terraform apply.

The init call is not repeated if any successful execution has happened already, to save time.

Returns:

Type	Description
str	The path to the stack YAML configuration file as a string.

Exceptions:

Type	Description
RuntimeError	if terraform init fails.

Source code in zenml/cli/stack_recipes.py

def apply(self) -> str:
    """Function to call terraform init and terraform apply.

    The init call is not repeated if any successful execution has
    happened already, to save time.

    Returns:
        The path to the stack YAML configuration file as a string.

    Raises:
        RuntimeError: if terraform init fails.
    """
    # this directory gets created after a successful init
    previous_run_dir = os.path.join(self.tf.working_dir, ".ignoreme")
    if fileio.exists(previous_run_dir):
        logger.info(
            "Terraform already initialized, "
            "terraform init will not be executed."
        )
    else:
        ret_code, _, _ = self.tf.init(capture_output=False)
        if ret_code != 0:
            raise RuntimeError("The command 'terraform init' failed.")
        fileio.mkdir(previous_run_dir)

    # get variables from the recipe as a python dictionary
    vars = self._get_vars(self.tf.working_dir)

    # once init is successful, call terraform apply
    self.tf.apply(
        var=vars,
        input=False,
        capture_output=False,
        raise_on_error=True,
    )

    # return the path of the stack yaml file
    _, stack_file_path, _ = self.tf.output("stack-yaml-path")
    return str(stack_file_path)

check_installation(self)

Checks if terraform is installed on the host system.

Exceptions:

Type	Description
RuntimeError	if terraform is not installed.

Source code in zenml/cli/stack_recipes.py

def check_installation(self) -> None:
    """Checks if terraform is installed on the host system.

    Raises:
        RuntimeError: if terraform is not installed.
    """
    if not self._is_installed():
        raise RuntimeError(
            "Terraform is required for stack recipes to run and was not "
            "found installed on your machine. Please visit "
            "https://learn.hashicorp.com/tutorials/terraform/install-cli "
            "to install it."
        )

destroy(self)

Function to call terraform destroy on the given path.

Source code in zenml/cli/stack_recipes.py

def destroy(self) -> None:
    """Function to call terraform destroy on the given path."""
    self.tf.destroy(
        capture_output=False,
        raise_on_error=True,
        force=python_terraform.IsNotFlagged,
    )

set_log_level(self, log_level)

Set TF_LOG env var to the log_level provided by the user.

Parameters:

Name	Type	Description	Default
log_level	str	One of TRACE, DEBUG, INFO, WARN or ERROR to set as the log level for terraform CLI.	required

Source code in zenml/cli/stack_recipes.py

def set_log_level(self, log_level: str) -> None:
    """Set TF_LOG env var to the log_level provided by the user.

    Args:
        log_level: One of TRACE, DEBUG, INFO, WARN or ERROR to set
            as the log level for terraform CLI.
    """
    os.environ["TF_LOG"] = log_level