Skip to content

Example

zenml.cli.example

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."""
        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.

__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."""
        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]:
        """In case a release branch is checked out, this property returns
        that version, else `None` is returned"""
        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."""
        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."""
        return self.cloning_path.exists()

    @property
    def examples_dir(self) -> str:
        """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."""
        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 delete(self) -> None:
        """Delete `cloning_path` if it exists."""
        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.

        Raises:
            GitCommandError: if branch doesn't exist.
        """
        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

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

examples_dir: str property readonly

Returns the path for the examples directory.

examples_run_bash_script: str property readonly

Path to the bash script that runs the example.

is_cloned: bool property readonly

Returns whether we have already cloned the examples repository.

latest_release_branch: str property readonly

Returns the name of the latest release branch.

__init__(self, cloning_path) special

Create a new ExamplesRepo instance.

Source code in zenml/cli/example.py 
def __init__(self, cloning_path: Path) -> None:
    """Create a new ExamplesRepo instance."""
    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.

Exceptions:

Type Description
GitCommandError

if branch doesn't exist.
Source code in zenml/cli/example.py 
def checkout(self, branch: str) -> None:
    """Checks out a specific branch or tag of the examples repository.

    Raises:
        GitCommandError: if branch doesn't exist.
    """
    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.

Source code in zenml/cli/example.py 
def delete(self) -> None:
    """Delete `cloning_path` if it exists."""
    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."
        )

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 = zenml.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."""
        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:
        """Returns a boolean 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"""
        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.
        """
        example_dict = {e.name: e for e in self.examples}
        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: {[example_dict.keys()]}"
                )
        else:
            return self.examples

    def pull(
        self,
        branch: str,
        force: bool = False,
    ) -> None:
        from git.exc import GitCommandError

        """Pulls the examples from the main git examples repository."""
        if not self.examples_repo.is_cloned:
            self.examples_repo.clone()
        elif force:
            self.examples_repo.delete()
            self.examples_repo.clone()

        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."""
        zenml.io.utils.create_dir_if_not_exists(destination_dir)
        zenml.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.

is_matching_versions: bool property readonly

Returns a boolean 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 = zenml.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.

Source code in zenml/cli/example.py 
def copy_example(self, example: Example, destination_dir: str) -> None:
    """Copies an example to the destination_dir."""
    zenml.io.utils.create_dir_if_not_exists(destination_dir)
    zenml.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
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.
    """
    example_dict = {e.name: e for e in self.examples}
    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: {[example_dict.keys()]}"
            )
    else:
        return self.examples

is_example(self, example_name=None)

Checks if 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"""
    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_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 all properties and methods of the local example that can be run from the CLI

Source code in zenml/cli/example.py 
class LocalExample:
    """Class to encapsulate all properties and methods of 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 drectl in local example directory
        the __init__.py file is excluded from this list"""
        py_in_dir = zenml.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"""
        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"""
        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"""
        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 executable_python_example(self) -> str:
        """Return the Python file for the example"""
        if 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."""
        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.
        """
        if all(map(fileio.exists, example_runner)):
            call = (
                example_runner
                + ["--executable", self.executable_python_example]
                + ["-f"] * 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

has_any_python_file: bool property readonly

Boolean that states if any python file is present

has_single_python_file: bool property readonly

Boolean that states if only one python file is present

python_files_in_dir: List[str] property readonly

List of all python files in the drectl in local example directory the init.py file is excluded from this list

run_dot_py_file: Optional[str] property readonly

Returns the 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.

Source code in zenml/cli/example.py 
def is_present(self) -> bool:
    """Checks if the example exists at the given path."""
    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
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.
    """
    if all(map(fileio.exists, example_runner)):
        call = (
            example_runner
            + ["--executable", self.executable_python_example]
            + ["-f"] * 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.

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."""
    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."
            )