Example

zenml.cli.example

Functionality to handle downloading ZenML examples via the CLI.

Example

Class for all example objects.

Source code in zenml/cli/example.py

class Example:
    """Class for all example objects."""

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

        Args:
            name: The name of the example, specifically the name of the folder
                  on git
            path_in_repo: Path to the local example 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 example.

        Returns:
            The README content associated with a particular example.

        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"Example {self.name} is not one of the available options."
                    f"\n"
                    f"To list all available examples, type: `zenml example "
                    f"list`"
                )

readme_content: str property readonly

Returns the README content associated with a particular example.

Returns:

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

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 Example instance.

Parameters:

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

Source code in zenml/cli/example.py

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

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

ExamplesRepo

Class for the examples repository object.

Source code in zenml/cli/example.py

class ExamplesRepo:
    """Class for the examples repository object."""

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

        Args:
            cloning_path: Path to the local examples 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 examples, "
                "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 examples."
            )
            self.clone()
            self.checkout_latest_release()

    @property
    def active_version(self) -> Optional[str]:
        """Returns the active version of the examples 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 examples 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
        )
        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 examples repository.

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

    @property
    def examples_dir(self) -> str:
        """Returns the path for the examples directory.

        Returns:
            The path for the examples directory.
        """
        return os.path.join(self.cloning_path, "examples")

    @property
    def examples_run_bash_script(self) -> str:
        """Path to the bash script that runs the example.

        Returns:
            Path to the bash script that runs the example.
        """
        return os.path.join(self.examples_dir, EXAMPLES_RUN_SCRIPT)

    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"Cloning repo {GIT_REPO_URL} to {self.cloning_path}")
            self.repo = Repo.clone_from(
                GIT_REPO_URL, self.cloning_path, branch="main"
            )
        except KeyboardInterrupt:
            self.delete()
            logger.error("Canceled download of repository.. Rolled back.")

    def pull(self) -> None:
        """Pull or clone the latest repository version."""
        if not self.cloning_path.exists():
            self.clone()
        else:
            logger.info("Fetching latest examples repository")
            self.repo.git.pull()

    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 examples 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 examples 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 examples repository."""
        self.checkout(branch=self.latest_release_branch)

active_version: Optional[str] property readonly

Returns the active version of the examples 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 examples repository.

examples_dir: str property readonly

Returns the path for the examples directory.

Returns:

Type	Description
str	The path for the examples directory.

examples_run_bash_script: str property readonly

Path to the bash script that runs the example.

Returns:

Type	Description
str	Path to the bash script that runs the example.

is_cloned: bool property readonly

Returns whether we have already cloned the examples repository.

Returns:

Type	Description
bool	Whether we have already cloned the examples 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 ExamplesRepo instance.

Parameters:

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

Exceptions:

Type	Description
GitNotFoundError	If git is not installed.

Source code in zenml/cli/example.py

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

    Args:
        cloning_path: Path to the local examples 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 examples, "
            "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 examples."
        )
        self.clone()
        self.checkout_latest_release()

checkout(self, branch)

Checks out a specific branch or tag of the examples repository.

Parameters:

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

Source code in zenml/cli/example.py

def checkout(self, branch: str) -> None:
    """Checks out a specific branch or tag of the examples 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 examples repository.

Source code in zenml/cli/example.py

def checkout_latest_release(self) -> None:
    """Checks out the latest release of the examples 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/example.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"Cloning repo {GIT_REPO_URL} to {self.cloning_path}")
        self.repo = Repo.clone_from(
            GIT_REPO_URL, self.cloning_path, branch="main"
        )
    except KeyboardInterrupt:
        self.delete()
        logger.error("Canceled download of repository.. 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/example.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 examples repository from "
            f"{self.cloning_path} as it does not exist."
        )

pull(self)

Pull or clone the latest repository version.

Source code in zenml/cli/example.py

def pull(self) -> None:
    """Pull or clone the latest repository version."""
    if not self.cloning_path.exists():
        self.clone()
    else:
        logger.info("Fetching latest examples repository")
        self.repo.git.pull()

GitExamplesHandler

Class for the GitExamplesHandler that interfaces with the CLI tool.

Source code in zenml/cli/example.py

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

    def __init__(self) -> None:
        """Create a new GitExamplesHandler instance."""
        self.repo_dir = io_utils.get_global_config_directory()
        self.examples_dir = Path(
            os.path.join(self.repo_dir, EXAMPLES_GITHUB_REPO)
        )
        self.examples_repo = ExamplesRepo(self.examples_dir)

    @property
    def examples(self) -> List[Example]:
        """Property that contains a list of examples.

        Returns:
            A list of examples.
        """
        return [
            Example(
                name, Path(os.path.join(self.examples_repo.examples_dir, name))
            )
            for name in sorted(os.listdir(self.examples_repo.examples_dir))
            if (
                not name.startswith(".")
                and not name.startswith("__")
                and not name.startswith("README")
                and not name.endswith(".sh")
            )
        ]

    @property
    def is_matching_versions(self) -> bool:
        """Whether the checked out examples are on the same code version as ZenML.

        Returns:
            Whether the checked out examples are on the same code version as ZenML.
        """
        return zenml_version_installed == str(self.examples_repo.active_version)

    def is_example(self, example_name: Optional[str] = None) -> bool:
        """Checks if the supplied example_name corresponds to an example.

        Args:
            example_name: The name of the example to check.

        Returns:
            Whether the supplied example_name corresponds to an example.
        """
        example_dict = {e.name: e for e in self.examples}
        if example_name:
            if example_name in example_dict.keys():
                return True

        return False

    def get_examples(self, example_name: Optional[str] = None) -> List[Example]:
        """Method that allows you to get an example by name.

        If no example is supplied,  all examples are returned.

        Args:
            example_name: Name of an example.

        Returns:
            A list of examples.

        Raises:
            KeyError: If the supplied example_name is not found.
        """
        example_dict = {
            e.name: e
            for e in self.examples
            if e.name not in EXCLUDED_EXAMPLE_DIRS
        }
        if example_name:
            if example_name in example_dict.keys():
                return [example_dict[example_name]]
            else:
                raise KeyError(
                    f"Example {example_name} does not exist! "
                    f"Available examples: {list(example_dict)}"
                )
        else:
            return self.examples

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

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

        if self.examples_repo.is_cloned and force:
            self.examples_repo.delete()
        self.examples_repo.pull()

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

    def pull_latest_examples(self) -> None:
        """Pulls the latest examples from the examples repository."""
        self.pull(branch=self.examples_repo.latest_release_branch, force=True)

    def copy_example(self, example: Example, destination_dir: str) -> None:
        """Copies an example to the destination_dir.

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

    def clean_current_examples(self) -> None:
        """Deletes the ZenML examples directory from your current working directory."""
        examples_directory = os.path.join(os.getcwd(), "zenml_examples")
        shutil.rmtree(examples_directory)

examples: List[zenml.cli.example.Example] property readonly

Property that contains a list of examples.

Returns:

Type	Description
List[zenml.cli.example.Example]	A list of examples.

is_matching_versions: bool property readonly

Whether the checked out examples are on the same code version as ZenML.

Returns:

Type	Description
bool	Whether the checked out examples are on the same code version as ZenML.

__init__(self) special

Create a new GitExamplesHandler instance.

Source code in zenml/cli/example.py

def __init__(self) -> None:
    """Create a new GitExamplesHandler instance."""
    self.repo_dir = io_utils.get_global_config_directory()
    self.examples_dir = Path(
        os.path.join(self.repo_dir, EXAMPLES_GITHUB_REPO)
    )
    self.examples_repo = ExamplesRepo(self.examples_dir)

clean_current_examples(self)

Deletes the ZenML examples directory from your current working directory.

Source code in zenml/cli/example.py

def clean_current_examples(self) -> None:
    """Deletes the ZenML examples directory from your current working directory."""
    examples_directory = os.path.join(os.getcwd(), "zenml_examples")
    shutil.rmtree(examples_directory)

copy_example(self, example, destination_dir)

Copies an example to the destination_dir.

Parameters:

Name	Type	Description	Default
example	Example	The example to copy.	required
destination_dir	str	The destination directory to copy the example to.	required

Source code in zenml/cli/example.py

def copy_example(self, example: Example, destination_dir: str) -> None:
    """Copies an example to the destination_dir.

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

get_examples(self, example_name=None)

Method that allows you to get an example by name.

If no example is supplied, all examples are returned.

Parameters:

Name	Type	Description	Default
example_name	Optional[str]	Name of an example.	None

Returns:

Type	Description
List[zenml.cli.example.Example]	A list of examples.

Exceptions:

Type	Description
KeyError	If the supplied example_name is not found.

Source code in zenml/cli/example.py

def get_examples(self, example_name: Optional[str] = None) -> List[Example]:
    """Method that allows you to get an example by name.

    If no example is supplied,  all examples are returned.

    Args:
        example_name: Name of an example.

    Returns:
        A list of examples.

    Raises:
        KeyError: If the supplied example_name is not found.
    """
    example_dict = {
        e.name: e
        for e in self.examples
        if e.name not in EXCLUDED_EXAMPLE_DIRS
    }
    if example_name:
        if example_name in example_dict.keys():
            return [example_dict[example_name]]
        else:
            raise KeyError(
                f"Example {example_name} does not exist! "
                f"Available examples: {list(example_dict)}"
            )
    else:
        return self.examples

is_example(self, example_name=None)

Checks if the supplied example_name corresponds to an example.

Parameters:

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

Returns:

Type	Description
bool	Whether the supplied example_name corresponds to an example.

Source code in zenml/cli/example.py

def is_example(self, example_name: Optional[str] = None) -> bool:
    """Checks if the supplied example_name corresponds to an example.

    Args:
        example_name: The name of the example to check.

    Returns:
        Whether the supplied example_name corresponds to an example.
    """
    example_dict = {e.name: e for e in self.examples}
    if example_name:
        if example_name in example_dict.keys():
            return True

    return False

pull(self, branch, force=False)

Pulls the examples from the main git examples 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/example.py

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

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

    if self.examples_repo.is_cloned and force:
        self.examples_repo.delete()
    self.examples_repo.pull()

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

pull_latest_examples(self)

Pulls the latest examples from the examples repository.

Source code in zenml/cli/example.py

def pull_latest_examples(self) -> None:
    """Pulls the latest examples from the examples repository."""
    self.pull(branch=self.examples_repo.latest_release_branch, force=True)

LocalExample

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

Source code in zenml/cli/example.py

class LocalExample:
    """Class to encapsulate the local example that can be run from the CLI."""

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

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

    @property
    def python_files_in_dir(self) -> List[str]:
        """List of all Python files in the local example directory.

        The `__init__.py` file is excluded from this list.

        Returns:
            List of Python files in the local example directory.
        """
        py_in_dir = io_utils.find_files(str(self.path), "*.py")
        py_files = []
        for file in py_in_dir:
            # Make sure only files directly in dir are considered, not files
            # in sub dirs
            if self.path == Path(file).parent:
                if Path(file).name != "__init__.py":
                    py_files.append(file)

        return py_files

    @property
    def has_single_python_file(self) -> bool:
        """Boolean that states if only one Python file is present.

        Returns:
            Whether only one Python file is present.
        """
        return len(self.python_files_in_dir) == 1

    @property
    def has_any_python_file(self) -> bool:
        """Boolean that states if any python file is present.

        Returns:
            Whether any Python file is present.
        """
        return len(self.python_files_in_dir) > 0

    @property
    def run_dot_py_file(self) -> Optional[str]:
        """Returns the path to the run.py file in case one exists.

        Returns:
            Path to the run.py file in case one exists.
        """
        for file in self.python_files_in_dir:
            # Make sure only files directly in dir are considered, not files
            # in sub dirs
            if self.path == Path(file).parent:
                if Path(file).name == "run.py":
                    return file
        return None

    @property
    def needs_manual_user_setup(self) -> bool:
        """Checks if a setup.sh file exists in the example dir.

        This indicates the possibility to run the example without any user
        input. Examples with no setup.sh file need the user to setup
        infrastructure and/or connect to tools/service providers.

        Returns:
            True if no setup.sh file in self.path, False else
        """
        return not fileio.exists(os.path.join(str(self.path), "setup.sh"))

    @property
    def executable_python_example(self) -> str:
        """Return the Python file for the example.

        Returns:
            The Python file for the example.

        Raises:
            RuntimeError: If no runner script is present in the example.
            NotImplementedError: If the examples needs manual user setup.
        """
        if self.needs_manual_user_setup:
            raise NotImplementedError(
                "This example currently does not support being run from the "
                "CLI as user specific setup is required. Consult the README.md "
                "of the example to find out more."
            )
        elif self.has_single_python_file:
            return self.python_files_in_dir[0]
        elif self.run_dot_py_file:
            return self.run_dot_py_file
        elif self.has_any_python_file:
            logger.warning(
                "This example has multiple executable python files. "
                "The last one in alphanumerical order is taken."
            )
            return sorted(self.python_files_in_dir)[-1]
        else:
            raise RuntimeError(
                "No pipeline runner script found in example. "
                f"Files found: {self.python_files_in_dir}"
            )

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

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

    def run_example(
        self,
        example_runner: List[str],
        force: bool,
        prevent_stack_setup: bool = False,
    ) -> None:
        """Run the local example using the bash script at the supplied location.

        Args:
            example_runner: Sequence of locations of executable file(s)
                            to run the example
            force: Whether to force the install
            prevent_stack_setup: Prevents the example from setting up a custom
                stack.

        Raises:
            NotImplementedError: If the example hasn't been implement yet.
            FileNotFoundError: If the example runner script is not found.
            subprocess.CalledProcessError: If the example runner script fails.
        """
        if all(map(fileio.exists, example_runner)):
            call = (
                example_runner
                + ["--executable", self.executable_python_example]
                + ["-y"] * force
                + ["--no-stack-setup"] * prevent_stack_setup
            )
            try:
                # TODO [ENG-271]: Catch errors that might be thrown
                #  in subprocess
                subprocess.check_call(
                    call,
                    cwd=str(self.path),
                    shell=click._compat.WIN,
                    env=os.environ.copy(),
                )
            except RuntimeError:
                raise NotImplementedError(
                    f"Currently the example {self.name} "
                    "has no implementation for the "
                    "run method"
                )
            except subprocess.CalledProcessError as e:
                if e.returncode == 38:
                    raise NotImplementedError(
                        f"Currently the example {self.name} "
                        "has no implementation for the "
                        "run method"
                    )
                raise
        else:
            raise FileNotFoundError(
                "Bash File(s) to run Examples not found at" f"{example_runner}"
            )

        # Telemetry
        track_event(AnalyticsEvent.RUN_EXAMPLE, {"example_name": self.name})

executable_python_example: str property readonly

Return the Python file for the example.

Returns:

Type	Description
str	The Python file for the example.

Exceptions:

Type	Description
RuntimeError	If no runner script is present in the example.
NotImplementedError	If the examples needs manual user setup.

has_any_python_file: bool property readonly

Boolean that states if any python file is present.

Returns:

Type	Description
bool	Whether any Python file is present.

has_single_python_file: bool property readonly

Boolean that states if only one Python file is present.

Returns:

Type	Description
bool	Whether only one Python file is present.

needs_manual_user_setup: bool property readonly

Checks if a setup.sh file exists in the example dir.

This indicates the possibility to run the example without any user input. Examples with no setup.sh file need the user to setup infrastructure and/or connect to tools/service providers.

Returns:

Type	Description
bool	True if no setup.sh file in self.path, False else

python_files_in_dir: List[str] property readonly

List of all Python files in the local example directory.

The __init__.py file is excluded from this list.

Returns:

Type	Description
List[str]	List of Python files in the local example directory.

run_dot_py_file: Optional[str] property readonly

Returns the path to the run.py file in case one exists.

Returns:

Type	Description
Optional[str]	Path to the run.py file in case one exists.

__init__(self, path, name) special

Create a new LocalExample instance.

Parameters:

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

Source code in zenml/cli/example.py

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

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

is_present(self)

Checks if the example exists at the given path.

Returns:

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

Source code in zenml/cli/example.py

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

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

run_example(self, example_runner, force, prevent_stack_setup=False)

Run the local example using the bash script at the supplied location.

Parameters:

Name	Type	Description	Default
example_runner	List[str]	Sequence of locations of executable file(s) to run the example	required
force	bool	Whether to force the install	required
prevent_stack_setup	bool	Prevents the example from setting up a custom stack.	False

Exceptions:

Type	Description
NotImplementedError	If the example hasn't been implement yet.
FileNotFoundError	If the example runner script is not found.
subprocess.CalledProcessError	If the example runner script fails.

Source code in zenml/cli/example.py

def run_example(
    self,
    example_runner: List[str],
    force: bool,
    prevent_stack_setup: bool = False,
) -> None:
    """Run the local example using the bash script at the supplied location.

    Args:
        example_runner: Sequence of locations of executable file(s)
                        to run the example
        force: Whether to force the install
        prevent_stack_setup: Prevents the example from setting up a custom
            stack.

    Raises:
        NotImplementedError: If the example hasn't been implement yet.
        FileNotFoundError: If the example runner script is not found.
        subprocess.CalledProcessError: If the example runner script fails.
    """
    if all(map(fileio.exists, example_runner)):
        call = (
            example_runner
            + ["--executable", self.executable_python_example]
            + ["-y"] * force
            + ["--no-stack-setup"] * prevent_stack_setup
        )
        try:
            # TODO [ENG-271]: Catch errors that might be thrown
            #  in subprocess
            subprocess.check_call(
                call,
                cwd=str(self.path),
                shell=click._compat.WIN,
                env=os.environ.copy(),
            )
        except RuntimeError:
            raise NotImplementedError(
                f"Currently the example {self.name} "
                "has no implementation for the "
                "run method"
            )
        except subprocess.CalledProcessError as e:
            if e.returncode == 38:
                raise NotImplementedError(
                    f"Currently the example {self.name} "
                    "has no implementation for the "
                    "run method"
                )
            raise
    else:
        raise FileNotFoundError(
            "Bash File(s) to run Examples not found at" f"{example_runner}"
        )

    # Telemetry
    track_event(AnalyticsEvent.RUN_EXAMPLE, {"example_name": self.name})

check_for_version_mismatch(git_examples_handler)

Prints a warning if the example version and ZenML version don't match.

Parameters:

Name	Type	Description	Default
git_examples_handler	GitExamplesHandler	The GitExamplesHandler instance.	required

Source code in zenml/cli/example.py

def check_for_version_mismatch(
    git_examples_handler: GitExamplesHandler,
) -> None:
    """Prints a warning if the example version and ZenML version don't match.

    Args:
        git_examples_handler: The GitExamplesHandler instance.
    """
    if git_examples_handler.is_matching_versions:
        return
    else:
        if git_examples_handler.examples_repo.active_version:
            warning(
                "The examples you have installed are installed with Version "
                f"{git_examples_handler.examples_repo.active_version} "
                f"of ZenML. However your code is at {zenml_version_installed} "
                "Consider using `zenml example pull` to download  "
                "examples matching your zenml installation."
            )
        else:
            warning(
                "The examples you have installed are downloaded from a "
                "development branch of ZenML. Full functionality is not "
                "guaranteed. Use `zenml example pull` to "
                "get examples using your zenml version."
            )