Utils

zenml.cli.utils

activate_integrations(func)

Decorator that activates all ZenML integrations.

Source code in zenml/cli/utils.py

def activate_integrations(func: F) -> F:
    """Decorator that activates all ZenML integrations."""

    @functools.wraps(func)
    def _wrapper(*args: Any, **kwargs: Any) -> Any:
        """Inner decorator function"""
        from zenml.integrations.registry import integration_registry

        integration_registry.activate_integrations()
        return func(*args, **kwargs)

    return cast(F, _wrapper)

confirmation(text, *args, **kwargs)

Echo a confirmation string on the CLI.

Parameters:

Name	Type	Description	Default
text	str	Input text string.	required
*args	Any	Args to be passed to click.confirm().	()
**kwargs	Any	Kwargs to be passed to click.confirm().	{}

Returns:

Type	Description
bool	Boolean based on user response.

Source code in zenml/cli/utils.py

def confirmation(text: str, *args: Any, **kwargs: Any) -> bool:
    """Echo a confirmation string on the CLI.

    Args:
      text: Input text string.
      *args: Args to be passed to click.confirm().
      **kwargs: Kwargs to be passed to click.confirm().

    Returns:
        Boolean based on user response.
    """
    return click.confirm(click.style(text, fg="yellow"), *args, **kwargs)

declare(text)

Echo a declaration on the CLI.

Parameters:

Name	Type	Description	Default
text	str	Input text string.	required

Source code in zenml/cli/utils.py

def declare(text: str) -> None:
    """Echo a declaration on the CLI.

    Args:
      text: Input text string.
    """
    click.echo(click.style(text, fg="green"))

error(text)

Echo an error string on the CLI.

Parameters:

Name	Type	Description	Default
text	str	Input text string.	required

Exceptions:

Type	Description
click.ClickException	when called.

Source code in zenml/cli/utils.py

def error(text: str) -> None:
    """Echo an error string on the CLI.

    Args:
      text: Input text string.

    Raises:
        click.ClickException: when called.
    """
    raise click.ClickException(message=click.style(text, fg="red", bold=True))

format_date(dt, format='%Y-%m-%d %H:%M:%S')

Format a date into a string.

Parameters:

Name	Type	Description	Default
dt	datetime	Datetime object to be formatted.	required
format	str	The format in string you want the datetime formatted to.	'%Y-%m-%d %H:%M:%S'

Returns:

Type	Description
str	Formatted string according to specification.

Source code in zenml/cli/utils.py

def format_date(
    dt: datetime.datetime, format: str = "%Y-%m-%d %H:%M:%S"
) -> str:
    """Format a date into a string.

    Args:
      dt: Datetime object to be formatted.
      format: The format in string you want the datetime formatted to.

    Returns:
        Formatted string according to specification.
    """
    if dt is None:
        return ""
    # make sure this is UTC
    dt = dt.replace(tzinfo=tz.tzutc())

    if sys.platform != "win32":
        # On non-windows get local time zone.
        local_zone = tz.tzlocal()
        dt = dt.astimezone(local_zone)
    else:
        logger.warning("On Windows, all times are displayed in UTC timezone.")

    return dt.strftime(format)

install_package(package)

Installs pypi package into the current environment with pip

Source code in zenml/cli/utils.py

def install_package(package: str) -> None:
    """Installs pypi package into the current environment with pip"""
    subprocess.check_call([sys.executable, "-m", "pip", "install", package])

parse_unknown_options(args)

Parse unknown options from the CLI.

Parameters:

Name	Type	Description	Default
args	List[str]	A list of strings from the CLI.	required

Returns:

Type	Description
Dict[str, Any]	Dict of parsed args.

Source code in zenml/cli/utils.py

def parse_unknown_options(args: List[str]) -> Dict[str, Any]:
    """Parse unknown options from the CLI.

    Args:
      args: A list of strings from the CLI.

    Returns:
        Dict of parsed args.
    """
    warning_message = (
        "Please provide args with a proper "
        "identifier as the key and the following structure: "
        '--custom_argument="value"'
    )

    assert all(a.startswith("--") for a in args), warning_message
    assert all(len(a.split("=")) == 2 for a in args), warning_message

    p_args = [a.lstrip("--").split("=") for a in args]

    assert all(k.isidentifier() for k, _ in p_args), warning_message

    r_args = {k: v for k, v in p_args}
    assert len(p_args) == len(r_args), "Replicated arguments!"

    return r_args

pretty_print(obj)

Pretty print an object on the CLI.

Parameters:

Name	Type	Description	Default
obj	Any	Any object with a str method defined.	required

Source code in zenml/cli/utils.py

def pretty_print(obj: Any) -> None:
    """Pretty print an object on the CLI.

    Args:
      obj: Any object with a __str__ method defined.
    """
    click.echo(str(obj))

print_stack_component_configuration(component)

Prints the configuration options of a stack component.

Source code in zenml/cli/utils.py

def print_stack_component_configuration(component: StackComponent) -> None:
    """Prints the configuration options of a stack component."""
    declare(f"NAME: {component.name}")
    for key, value in component.dict(exclude={"name"}).items():
        declare(f"{key.upper()}: {value}")

print_stack_component_list(components, active_component_name)

Prints a table with configuration options for a list of stack components.

If a component is active (its name matches the active_component_name), it will be highlighted in a separate table column.

Parameters:

Name	Type	Description	Default
components	List[zenml.stack.stack_component.StackComponent]	List of stack components to print.	required
active_component_name	str	Name of the component that is currently active.	required

Source code in zenml/cli/utils.py

def print_stack_component_list(
    components: List[StackComponent], active_component_name: str
) -> None:
    """Prints a table with configuration options for a list of stack components.

    If a component is active (its name matches the `active_component_name`),
    it will be highlighted in a separate table column.

    Args:
        components: List of stack components to print.
        active_component_name: Name of the component that is currently active.
    """
    configurations = []
    for component in components:
        is_active = component.name == active_component_name
        component_config = {
            "ACTIVE": "*" if is_active else "",
            **{key.upper(): value for key, value in component.dict().items()},
        }
        configurations.append(component_config)

    print_table(configurations)

print_table(obj)

Echoes the list of dicts in a table format. The input object should be a List of Dicts. Each item in that list represent a line in the Table. Each dict should have the same keys. The keys of the dict will be used as headers of the resulting table.

Parameters:

Name	Type	Description	Default
obj	List[Dict[str, Any]]	A List containing dictionaries.	required

Source code in zenml/cli/utils.py

def print_table(obj: List[Dict[str, Any]]) -> None:
    """Echoes the list of dicts in a table format. The input object should be a
    List of Dicts. Each item in that list represent a line in the Table. Each
    dict should have the same keys. The keys of the dict will be used as
    headers of the resulting table.

    Args:
      obj: A List containing dictionaries.
    """
    click.echo(tabulate(obj, headers="keys"))

title(text)

Echo a title formatted string on the CLI.

Parameters:

Name	Type	Description	Default
text	str	Input text string.	required

Source code in zenml/cli/utils.py

def title(text: str) -> None:
    """Echo a title formatted string on the CLI.

    Args:
      text: Input text string.
    """
    click.echo(click.style(text.upper(), fg="cyan", bold=True, underline=True))

uninstall_package(package)

Uninstalls pypi package from the current environment with pip

Source code in zenml/cli/utils.py

def uninstall_package(package: str) -> None:
    """Uninstalls pypi package from the current environment with pip"""
    subprocess.check_call(
        [sys.executable, "-m", "pip", "uninstall", "-y", package]
    )

warning(text)

Echo a warning string on the CLI.

Parameters:

Name	Type	Description	Default
text	str	Input text string.	required

Source code in zenml/cli/utils.py

def warning(text: str) -> None:
    """Echo a warning string on the CLI.

    Args:
      text: Input text string.
    """
    click.echo(click.style(text, fg="yellow", bold=True))