Environment

zenml.environment

Environment implementation.

Attributes

INSIDE_ZENML_CONTAINER = handle_bool_env_var(ENV_ZENML_CONTAINER, False) module-attribute

logger = get_logger(__name__) module-attribute

Classes

Environment()

Provides environment information.

Initializes an Environment instance.

Note: Environment is a singleton class, which means this method will only get called once. All following Environment() calls will return the previously initialized instance.

Source code in src/zenml/environment.py

def __init__(self) -> None:
    """Initializes an Environment instance.

    Note: Environment is a singleton class, which means this method will
    only get called once. All following `Environment()` calls will return
    the previously initialized instance.
    """

Functions

get_python_packages() -> List[str] staticmethod

Returns a list of installed Python packages.

Raises:

Type	Description
RuntimeError	If the process to get the list of installed packages fails.

Returns:

Type	Description
List[str]	List of installed packages in pip freeze format.

Source code in src/zenml/environment.py

@staticmethod
def get_python_packages() -> List[str]:
    """Returns a list of installed Python packages.

    Raises:
        RuntimeError: If the process to get the list of installed packages
            fails.

    Returns:
        List of installed packages in pip freeze format.
    """
    try:
        output = subprocess.check_output(["pip", "freeze"]).decode()
        return output.strip().split("\n")
    except subprocess.CalledProcessError:
        raise RuntimeError(
            "Failed to get list of installed Python packages"
        )

get_system_info() -> Dict[str, str] staticmethod

Information about the operating system.

Returns:

Type	Description
Dict[str, str]	A dictionary containing information about the operating system.

Source code in src/zenml/environment.py

@staticmethod
def get_system_info() -> Dict[str, str]:
    """Information about the operating system.

    Returns:
        A dictionary containing information about the operating system.
    """
    system = platform.system()

    if system == "Windows":
        release, version, csd, ptype = platform.win32_ver()

        return {
            "os": "windows",
            "windows_version_release": release,
            "windows_version": version,
            "windows_version_service_pack": csd,
            "windows_version_os_type": ptype,
        }

    if system == "Darwin":
        return {"os": "mac", "mac_version": platform.mac_ver()[0]}

    if system == "Linux":
        return {
            "os": "linux",
            "linux_distro": distro.id(),
            "linux_distro_like": distro.like(),
            "linux_distro_version": distro.version(),
        }

    # We don't collect data for any other system.
    return {"os": "unknown"}

in_bitbucket_ci() -> bool staticmethod

If the current Python process is running in Bitbucket CI.

Returns:

Type	Description
bool	True if the current Python process is running in Bitbucket
bool	CI, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_bitbucket_ci() -> bool:
    """If the current Python process is running in Bitbucket CI.

    Returns:
        `True` if the current Python process is running in Bitbucket
        CI, `False` otherwise.
    """
    return "BITBUCKET_BUILD_NUMBER" in os.environ

in_ci() -> bool staticmethod

If the current Python process is running in any CI.

Returns:

Type	Description
bool	True if the current Python process is running in any
bool	CI, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_ci() -> bool:
    """If the current Python process is running in any CI.

    Returns:
        `True` if the current Python process is running in any
        CI, `False` otherwise.
    """
    return "CI" in os.environ

in_circle_ci() -> bool staticmethod

If the current Python process is running in Circle CI.

Returns:

Type	Description
bool	True if the current Python process is running in Circle
bool	CI, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_circle_ci() -> bool:
    """If the current Python process is running in Circle CI.

    Returns:
        `True` if the current Python process is running in Circle
        CI, `False` otherwise.
    """
    return "CIRCLECI" in os.environ

in_container() -> bool staticmethod

If the current python process is running in a container.

Returns:

Type	Description
bool	True if the current python process is running in a
bool	container, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_container() -> bool:
    """If the current python process is running in a container.

    Returns:
        `True` if the current python process is running in a
        container, `False` otherwise.
    """
    # TODO [ENG-167]: Make this more reliable and add test.
    return INSIDE_ZENML_CONTAINER

in_docker() -> bool staticmethod

If the current python process is running in a docker container.

Returns:

Type	Description
bool	True if the current python process is running in a docker
bool	container, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_docker() -> bool:
    """If the current python process is running in a docker container.

    Returns:
        `True` if the current python process is running in a docker
        container, `False` otherwise.
    """
    if os.path.exists("./dockerenv") or os.path.exists("/.dockerinit"):
        return True

    try:
        with open("/proc/1/cgroup", "rt") as ifh:
            info = ifh.read()
            return "docker" in info
    except (FileNotFoundError, Exception):
        return False

in_github_actions() -> bool staticmethod

If the current Python process is running in GitHub Actions.

Returns:

Type	Description
bool	True if the current Python process is running in GitHub
bool	Actions, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_github_actions() -> bool:
    """If the current Python process is running in GitHub Actions.

    Returns:
        `True` if the current Python process is running in GitHub
        Actions, `False` otherwise.
    """
    return "GITHUB_ACTIONS" in os.environ

in_github_codespaces() -> bool staticmethod

If the current Python process is running in GitHub Codespaces.

Returns:

Type	Description
bool	True if the current Python process is running in GitHub Codespaces,
bool	False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_github_codespaces() -> bool:
    """If the current Python process is running in GitHub Codespaces.

    Returns:
        `True` if the current Python process is running in GitHub Codespaces,
        `False` otherwise.
    """
    return (
        "CODESPACES" in os.environ
        or "GITHUB_CODESPACE_TOKEN" in os.environ
        or "GITHUB_CODESPACES_PORT_FORWARDING_DOMAIN" in os.environ
    )

in_gitlab_ci() -> bool staticmethod

If the current Python process is running in GitLab CI.

Returns:

Type	Description
bool	True if the current Python process is running in GitLab
bool	CI, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_gitlab_ci() -> bool:
    """If the current Python process is running in GitLab CI.

    Returns:
        `True` if the current Python process is running in GitLab
        CI, `False` otherwise.
    """
    return "GITLAB_CI" in os.environ

in_google_colab() -> bool staticmethod

If the current Python process is running in a Google Colab.

Returns:

Type	Description
bool	True if the current Python process is running in a Google Colab,
bool	False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_google_colab() -> bool:
    """If the current Python process is running in a Google Colab.

    Returns:
        `True` if the current Python process is running in a Google Colab,
        `False` otherwise.
    """
    try:
        import google.colab  # noqa

        return True

    except ModuleNotFoundError:
        return False

in_kubernetes() -> bool staticmethod

If the current python process is running in a kubernetes pod.

Returns:

Type	Description
bool	True if the current python process is running in a kubernetes
bool	pod, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_kubernetes() -> bool:
    """If the current python process is running in a kubernetes pod.

    Returns:
        `True` if the current python process is running in a kubernetes
        pod, `False` otherwise.
    """
    if "KUBERNETES_SERVICE_HOST" in os.environ:
        return True

    try:
        with open("/proc/1/cgroup", "rt") as ifh:
            info = ifh.read()
            return "kubepod" in info
    except (FileNotFoundError, Exception):
        return False

in_lightning_ai_studio() -> bool staticmethod

If the current Python process is running in Lightning.ai studios.

Returns:

Type	Description
bool	True if the current Python process is running in Lightning.ai studios,
bool	False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_lightning_ai_studio() -> bool:
    """If the current Python process is running in Lightning.ai studios.

    Returns:
        `True` if the current Python process is running in Lightning.ai studios,
        `False` otherwise.
    """
    return (
        "LIGHTNING_CLOUD_URL" in os.environ
        and "LIGHTNING_CLOUDSPACE_HOST" in os.environ
    )

in_notebook() -> bool staticmethod

If the current Python process is running in a notebook.

Returns:

Type	Description
bool	True if the current Python process is running in a notebook,
bool	False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_notebook() -> bool:
    """If the current Python process is running in a notebook.

    Returns:
        `True` if the current Python process is running in a notebook,
        `False` otherwise.
    """
    if Environment.in_google_colab():
        return True

    try:
        ipython = get_ipython()  # type: ignore[name-defined]
    except NameError:
        return False

    if ipython.__class__.__name__ in [
        "TerminalInteractiveShell",
        "ZMQInteractiveShell",
        "DatabricksShell",
    ]:
        return True
    return False

in_paperspace_gradient() -> bool staticmethod

If the current Python process is running in Paperspace Gradient.

Returns:

Type	Description
bool	True if the current Python process is running in Paperspace
bool	Gradient, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_paperspace_gradient() -> bool:
    """If the current Python process is running in Paperspace Gradient.

    Returns:
        `True` if the current Python process is running in Paperspace
        Gradient, `False` otherwise.
    """
    return "PAPERSPACE_NOTEBOOK_REPO_ID" in os.environ

in_vscode_remote_container() -> bool staticmethod

If the current Python process is running in a VS Code Remote Container.

Returns:

Type	Description
bool	True if the current Python process is running in a VS Code Remote Container,
bool	False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_vscode_remote_container() -> bool:
    """If the current Python process is running in a VS Code Remote Container.

    Returns:
        `True` if the current Python process is running in a VS Code Remote Container,
        `False` otherwise.
    """
    return (
        "REMOTE_CONTAINERS" in os.environ
        or "VSCODE_REMOTE_CONTAINERS_SESSION" in os.environ
    )

in_wsl() -> bool staticmethod

If the current process is running in Windows Subsystem for Linux.

source: https://www.scivision.dev/python-detect-wsl/

Returns:

Type	Description
bool	True if the current process is running in WSL, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_wsl() -> bool:
    """If the current process is running in Windows Subsystem for Linux.

    source: https://www.scivision.dev/python-detect-wsl/

    Returns:
        `True` if the current process is running in WSL, `False` otherwise.
    """
    return "microsoft-standard" in platform.uname().release

in_zenml_codespace() -> bool staticmethod

If the current Python process is running in ZenML Codespaces.

Returns:

Type	Description
bool	True if the current Python process is running in ZenML Codespaces,
bool	False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_zenml_codespace() -> bool:
    """If the current Python process is running in ZenML Codespaces.

    Returns:
        `True` if the current Python process is running in ZenML Codespaces,
        `False` otherwise.
    """
    return os.environ.get("ZENML_ENVIRONMENT") == "codespace"

python_version() -> str staticmethod

Returns the python version of the running interpreter.

Returns:

Name	Type	Description
str	str	the python version

Source code in src/zenml/environment.py

@staticmethod
def python_version() -> str:
    """Returns the python version of the running interpreter.

    Returns:
        str: the python version
    """
    return platform.python_version()

EnvironmentType

Bases: StrEnum

Enum for environment types.

SingletonMetaClass(*args: Any, **kwargs: Any)

Bases: type

Singleton metaclass.

Use this metaclass to make any class into a singleton class:

class OneRing(metaclass=SingletonMetaClass):
    def __init__(self, owner):
        self._owner = owner

    @property
    def owner(self):
        return self._owner

the_one_ring = OneRing('Sauron')
the_lost_ring = OneRing('Frodo')
print(the_lost_ring.owner)  # Sauron
OneRing._clear() # ring destroyed

Initialize a singleton class.

Parameters:

Name	Type	Description	Default
*args	Any	Additional arguments.	()
**kwargs	Any	Additional keyword arguments.	{}

Source code in src/zenml/utils/singleton.py

def __init__(cls, *args: Any, **kwargs: Any) -> None:
    """Initialize a singleton class.

    Args:
        *args: Additional arguments.
        **kwargs: Additional keyword arguments.
    """
    super().__init__(*args, **kwargs)
    cls.__singleton_instance: Optional["SingletonMetaClass"] = None

Functions

Functions

get_environment() -> str

Returns a string representing the execution environment of the pipeline.

Returns:

Name	Type	Description
str	str	the execution environment

Source code in src/zenml/environment.py

def get_environment() -> str:
    """Returns a string representing the execution environment of the pipeline.

    Returns:
        str: the execution environment
    """
    # Order is important here
    if Environment.in_kubernetes():
        return EnvironmentType.KUBERNETES
    elif Environment.in_github_actions():
        return EnvironmentType.GITHUB_ACTION
    elif Environment.in_gitlab_ci():
        return EnvironmentType.GITLAB_CI
    elif Environment.in_circle_ci():
        return EnvironmentType.CIRCLE_CI
    elif Environment.in_bitbucket_ci():
        return EnvironmentType.BITBUCKET_CI
    elif Environment.in_ci():
        return EnvironmentType.GENERIC_CI
    elif Environment.in_github_codespaces():
        return EnvironmentType.GITHUB_CODESPACES
    elif Environment.in_zenml_codespace():
        return EnvironmentType.ZENML_CODESPACE
    elif Environment.in_vscode_remote_container():
        return EnvironmentType.VSCODE_REMOTE_CONTAINER
    elif Environment.in_lightning_ai_studio():
        return EnvironmentType.LIGHTNING_AI_STUDIO
    elif Environment.in_docker():
        return EnvironmentType.DOCKER
    elif Environment.in_container():
        return EnvironmentType.CONTAINER
    elif Environment.in_google_colab():
        return EnvironmentType.COLAB
    elif Environment.in_paperspace_gradient():
        return EnvironmentType.PAPERSPACE
    elif Environment.in_notebook():
        return EnvironmentType.NOTEBOOK
    elif Environment.in_wsl():
        return EnvironmentType.WSL
    else:
        return EnvironmentType.NATIVE

get_logger(logger_name: str) -> logging.Logger

Main function to get logger name,.

Parameters:

Name	Type	Description	Default
logger_name	str	Name of logger to initialize.	required

Returns:

Type	Description
Logger	A logger object.

Source code in src/zenml/logger.py

def get_logger(logger_name: str) -> logging.Logger:
    """Main function to get logger name,.

    Args:
        logger_name: Name of logger to initialize.

    Returns:
        A logger object.
    """
    logger = logging.getLogger(logger_name)
    logger.setLevel(get_logging_level().value)
    logger.addHandler(get_console_handler())

    logger.propagate = False
    return logger

get_run_environment_dict() -> Dict[str, Any]

Returns a dictionary of the current run environment.

Everything that is returned here will be saved in the DB as pipeline_run.client_environment and pipeline_run.orchestrator_environment for client and orchestrator respectively.

Returns:

Type	Description
Dict[str, Any]	A dictionary of the current run environment.

Source code in src/zenml/environment.py

def get_run_environment_dict() -> Dict[str, Any]:
    """Returns a dictionary of the current run environment.

    Everything that is returned here will be saved in the DB as
    `pipeline_run.client_environment` and
    `pipeline_run.orchestrator_environment` for client and orchestrator
    respectively.

    Returns:
        A dictionary of the current run environment.
    """
    env_dict: Dict[str, Any] = {
        "environment": str(get_environment()),
        **Environment.get_system_info(),
        "python_version": Environment.python_version(),
    }

    try:
        python_packages = Environment.get_python_packages()
    except RuntimeError:
        logger.warning("Failed to get list of installed Python packages")
    else:
        # TODO: We send the python packages as a string right now to keep
        # backwards compatibility with old versions. We should update this to
        # be a list of strings eventually.
        env_dict["python_packages"] = "\n".join(python_packages)

    return env_dict