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.file_exists(str(self.path_in_repo)) and fileio.is_dir(
                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:
            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:
            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:
        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:
        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:
        """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."""
        fileio.create_dir_if_not_exists(destination_dir)
        fileio.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."""
    fileio.create_dir_if_not_exists(destination_dir)
    fileio.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(self, branch, force=False)

Pulls the examples from the main git examples repository.

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

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 = fileio.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 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.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 is installed at the given path."""
        return fileio.file_exists(str(self.path)) and fileio.is_dir(
            str(self.path)
        )

    def run_example(self, example_runner: List[str], force: bool) -> 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
        """
        if all(map(fileio.file_exists, example_runner)):
            call = (
                example_runner
                + ["--executable", self.executable_python_example]
                + ["-f"] * force
            )
            try:
                # TODO [ENG-271]: Catch errors that might be thrown
                #  in subprocess
                subprocess.check_call(
                    call, cwd=str(self.path), shell=click._compat.WIN
                )
            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"
                    )
        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

__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 is installed at the given path.

Source code in zenml/cli/example.py 
def is_present(self) -> bool:
    """Checks if the example is installed at the given path."""
    return fileio.file_exists(str(self.path)) and fileio.is_dir(
        str(self.path)
    )

run_example(self, example_runner, force)

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
Source code in zenml/cli/example.py 
def run_example(self, example_runner: List[str], force: bool) -> 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
    """
    if all(map(fileio.file_exists, example_runner)):
        call = (
            example_runner
            + ["--executable", self.executable_python_example]
            + ["-f"] * force
        )
        try:
            # TODO [ENG-271]: Catch errors that might be thrown
            #  in subprocess
            subprocess.check_call(
                call, cwd=str(self.path), shell=click._compat.WIN
            )
        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"
                )
    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."
            )