Utils

zenml.cli.utils

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 Confirm.ask(text, console=console)
    return click.confirm(click.style(text, fg="yellow"), *args, **kwargs)

declare(text)

Echo a declaration on the CLI.

Parameters:

Name	Type	Description	Default
text	Union[str, rich.text.Text]	Input text string.	required

Source code in zenml/cli/utils.py

def declare(text: Union[str, Text]) -> None:
    """Echo a declaration on the CLI.

    Args:
      text: Input text string.
    """
    console.print(text, style="info")

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))
    # console.print(text, style="error")

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)

format_integration_list(integrations)

Formats a list of integrations into a List of Dicts. This list of dicts can then be printed in a table style using cli_utils.print_table.

Source code in zenml/cli/utils.py

def format_integration_list(
    integrations: List[Tuple[str, "IntegrationMeta"]]
) -> List[Dict[str, str]]:
    """Formats a list of integrations into a List of Dicts. This list of dicts
    can then be printed in a table style using cli_utils.print_table."""
    list_of_dicts = []
    for name, integration_impl in integrations:
        is_installed = integration_impl.check_installation()  # type: ignore[attr-defined]
        list_of_dicts.append(
            {
                "INSTALLED": ":white_check_mark:" if is_installed else "",
                "INTEGRATION": name,
                "REQUIRED_PACKAGES": ", ".join(integration_impl.REQUIREMENTS),  # type: ignore[attr-defined]
            }
        )
    return list_of_dicts

get_service_status_emoji(service)

Get the rich emoji representing the operational status of a Service.

Parameters:

Name	Type	Description	Default
service	BaseService	Service to get emoji for.	required

Returns:

Type	Description
str	String representing the emoji.

Source code in zenml/cli/utils.py

def get_service_status_emoji(service: BaseService) -> str:
    """Get the rich emoji representing the operational status of a Service.

    Args:
        service: Service to get emoji for.

    Returns:
        String representing the emoji.
    """
    if service.status.state == ServiceState.ACTIVE:
        return ":white_check_mark:"
    if service.status.state == ServiceState.INACTIVE:
        return ":pause_button:"
    if service.status.state == ServiceState.ERROR:
        return ":heavy_exclamation_mark:"
    return ":hourglass_not_done:"

install_packages(packages)

Installs pypi packages into the current environment with pip

Source code in zenml/cli/utils.py

def install_packages(packages: List[str]) -> None:
    """Installs pypi packages into the current environment with pip"""
    command = [
        sys.executable,
        "-m",
        "pip",
        "install",
    ] + packages

    if not IS_DEBUG_ENV:
        command += [
            "-qqq",
            "--no-warn-conflicts",
        ]

    subprocess.check_call(command)

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 using rich.print.

Parameters:

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

TODO: [LOW] check whether this needs to be converted to a string first

TODO: [LOW] use rich prettyprint for this instead

Source code in zenml/cli/utils.py

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

    Args:
      obj: Any object with a __str__ method defined.
    # TODO: [LOW] check whether this needs to be converted to a string first
    # TODO: [LOW] use rich prettyprint for this instead
    """
    console.print(obj)

pretty_print_model_deployer(model_services, model_deployer)

Given a list of served_models print all key value pairs associated with the secret

Parameters:

Name	Type	Description	Default
model_services	List[zenml.services.service.BaseService]	list of model deployment services	required
model_deployer	BaseModelDeployer	Active model deployer	required

Source code in zenml/cli/utils.py

def pretty_print_model_deployer(
    model_services: List[BaseService], model_deployer: BaseModelDeployer
) -> None:
    """Given a list of served_models print all key value pairs associated with
    the secret

    Args:
        model_services: list of model deployment services
        model_deployer: Active model deployer
    """
    model_service_dicts = []
    for model_service in model_services:
        served_model_info = model_deployer.get_model_server_info(model_service)

        model_service_dicts.append(
            {
                "STATUS": get_service_status_emoji(model_service),
                "UUID": str(model_service.uuid),
                "PIPELINE_NAME": model_service.config.pipeline_name,
                "PIPELINE_STEP_NAME": model_service.config.pipeline_step_name,
                "MODEL_NAME": served_model_info.get("MODEL_NAME", ""),
            }
        )

    print_table(model_service_dicts)

pretty_print_secret(secret, hide_secret=True)

Given a secret set print all key value pairs associated with the secret

Parameters:

Name	Type	Description	Default
secret	BaseSecretSchema	Secret of type BaseSecretSchema	required
hide_secret	bool	boolean that configures if the secret values are shown on the CLI	True

Source code in zenml/cli/utils.py

def pretty_print_secret(
    secret: BaseSecretSchema, hide_secret: bool = True
) -> None:
    """Given a secret set print all key value pairs associated with the secret

    Args:
        secret: Secret of type BaseSecretSchema
        hide_secret: boolean that configures if the secret values are shown
            on the CLI
    """

    def get_secret_value(value: Any) -> str:
        if value is None:
            return ""
        if hide_secret:
            return "***"
        return str(value)

    stack_dicts = [
        {
            "SECRET_NAME": secret.name,
            "SECRET_KEY": key,
            "SECRET_VALUE": get_secret_value(value),
        }
        for key, value in secret.content.items()
    ]
    print_table(stack_dicts)

print_active_profile()

Print active profile.

Source code in zenml/cli/utils.py

def print_active_profile() -> None:
    """Print active profile."""
    repo = Repository()
    scope = "local" if repo.root else "global"
    declare(
        f"Running with active profile: '{repo.active_profile_name}' ({scope})"
    )

print_active_stack()

Print active stack.

Source code in zenml/cli/utils.py

def print_active_stack() -> None:
    """Print active stack."""
    repo = Repository()
    declare(f"Running with active stack: '{repo.active_stack_name}'")

print_profile(profile, active)

Prints the configuration options of a profile.

Parameters:

Name	Type	Description	Default
profile	ProfileConfiguration	Profile to print.	required
active	bool	Whether the profile is active.	required
name		Name of the profile.	required

Source code in zenml/cli/utils.py

def print_profile(
    profile: ProfileConfiguration,
    active: bool,
) -> None:
    """Prints the configuration options of a profile.

    Args:
        profile: Profile to print.
        active: Whether the profile is active.
        name: Name of the profile.
    """
    profile_title = f"'{profile.name}' Profile Configuration"
    if active:
        profile_title += " (ACTIVE)"

    rich_table = table.Table(
        box=box.HEAVY_EDGE,
        title=profile_title,
        show_lines=True,
    )
    rich_table.add_column("PROPERTY")
    rich_table.add_column("VALUE")
    items = profile.dict().items()
    for item in items:
        rich_table.add_row(*[str(elem) for elem in item])

    # capitalize entries in first column
    rich_table.columns[0]._cells = [
        component.upper() for component in rich_table.columns[0]._cells  # type: ignore[union-attr]
    ]
    console.print(rich_table)

print_pydantic_models(models)

Prints the list of Pydantic models in a table.

Parameters:

Name	Type	Description	Default
models	Sequence[pydantic.main.BaseModel]	List of pydantic models that will be represented as a row in the table.	required

Source code in zenml/cli/utils.py

def print_pydantic_models(models: Sequence[BaseModel]) -> None:
    """Prints the list of Pydantic models in a table.

    Args:
        models: List of pydantic models that will be represented as a row in
            the table.
    """
    model_dicts = [
        {key: str(value) for key, value in model.dict().items()}
        for model in models
    ]
    print_table(model_dicts)

print_secrets(secrets)

Prints the configuration options of a stack.

Parameters:

Name	Type	Description	Default
secrets	List[str]	List of secrets	required

Source code in zenml/cli/utils.py

def print_secrets(secrets: List[str]) -> None:
    """Prints the configuration options of a stack.

    Args:
        secrets: List of secrets
    """
    rich_table = table.Table(
        box=box.HEAVY_EDGE,
        title="Secrets",
        show_lines=True,
    )
    rich_table.add_column("SECRET_NAME")
    secrets.sort()
    for item in secrets:
        rich_table.add_row(item)

    console.print(rich_table)

print_served_model_configuration(model_service, model_deployer)

Prints the configuration of a model_service.

Parameters:

Name	Type	Description	Default
model_service	BaseService	Specific service instance to	required
model_deployer	BaseModelDeployer	Active model deployer	required

Source code in zenml/cli/utils.py

def print_served_model_configuration(
    model_service: BaseService, model_deployer: BaseModelDeployer
) -> None:
    """Prints the configuration of a model_service.

    Args:
        model_service: Specific service instance to
        model_deployer: Active model deployer
    """
    title = f"Properties of Served Model {model_service.uuid}"

    rich_table = table.Table(
        box=box.HEAVY_EDGE,
        title=title,
        show_lines=True,
    )
    rich_table.add_column("MODEL SERVICE PROPERTY")
    rich_table.add_column("VALUE")

    # Get implementation specific info
    served_model_info = model_deployer.get_model_server_info(model_service)

    served_model_info = {
        **served_model_info,
        "UUID": str(model_service.uuid),
        "STATUS": get_service_status_emoji(model_service),
        "STATUS_MESSAGE": model_service.status.last_error,
        "PIPELINE_NAME": model_service.config.pipeline_name,
        "PIPELINE_RUN_ID": model_service.config.pipeline_run_id,
        "PIPELINE_STEP_NAME": model_service.config.pipeline_step_name,
    }

    # Sort fields alphabetically
    sorted_items = {k: v for k, v in sorted(served_model_info.items())}

    for item in sorted_items.items():
        rich_table.add_row(*[str(elem) for elem in item])

    # capitalize entries in first column
    rich_table.columns[0]._cells = [
        component.upper()  # type: ignore[union-attr]
        for component in rich_table.columns[0]._cells
    ]
    console.print(rich_table)

print_stack_component_configuration(component, display_name, active_status)

Prints the configuration options of a stack component.

Source code in zenml/cli/utils.py

def print_stack_component_configuration(
    component: StackComponent, display_name: str, active_status: bool
) -> None:
    """Prints the configuration options of a stack component."""
    title = f"{component.TYPE.value.upper()} Component Configuration"
    if active_status:
        title += " (ACTIVE)"
    rich_table = table.Table(
        box=box.HEAVY_EDGE,
        title=title,
        show_lines=True,
    )
    rich_table.add_column("COMPONENT_PROPERTY")
    rich_table.add_column("VALUE")
    items = component.dict().items()
    for item in items:
        rich_table.add_row(*[str(elem) for elem in item])

    # capitalize entries in first column
    rich_table.columns[0]._cells = [
        component.upper() for component in rich_table.columns[0]._cells  # type: ignore[union-attr]
    ]
    console.print(rich_table)

print_stack_component_list(components, active_component_name=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.

Parameters:

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

Source code in zenml/cli/utils.py

def print_stack_component_list(
    components: List[StackComponent],
    active_component_name: Optional[str] = None,
) -> 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": ":point_right:" if is_active else "",
            **{
                key.upper(): str(value)
                for key, value in component.dict().items()
            },
        }
        configurations.append(component_config)
    print_table(configurations)

print_stack_configuration(config, active, stack_name)

Prints the configuration options of a stack.

Source code in zenml/cli/utils.py

def print_stack_configuration(
    config: Dict[StackComponentType, str], active: bool, stack_name: str
) -> None:
    """Prints the configuration options of a stack."""
    stack_caption = f"'{stack_name}' stack"
    if active:
        stack_caption += " (ACTIVE)"
    rich_table = table.Table(
        box=box.HEAVY_EDGE,
        title="Stack Configuration",
        caption=stack_caption,
        show_lines=True,
    )
    rich_table.add_column("COMPONENT_TYPE")
    rich_table.add_column("COMPONENT_NAME")
    for component_type, name in config.items():
        rich_table.add_row(component_type.value, name)

    # capitalize entries in first column
    rich_table.columns[0]._cells = [
        component.upper() for component in rich_table.columns[0]._cells  # type: ignore[union-attr]
    ]
    console.print(rich_table)

print_table(obj)

Prints 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:
    """Prints 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.
    """
    column_keys = {key: None for dict_ in obj for key in dict_}
    column_names = [key.upper() for key in column_keys]
    rich_table = table.Table(*column_names, box=box.HEAVY_EDGE)

    for dict_ in obj:
        values = [dict_.get(key) for key in column_keys]
        rich_table.add_row(*values)

    if len(rich_table.columns) > 1:
        rich_table.columns[0].justify = "center"
    console.print(rich_table)

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.
    """
    console.print(text.upper(), style="title")

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",
            "-qqq",
            "-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.
    """
    console.print(text, style="warning")