content: #  Copyright (c) ZenML GmbH 2020. All Rights Reserved.
  #
  #  Licensed under the Apache License, Version 2.0 (the "License");
  #  you may not use this file except in compliance with the License.
  #  You may obtain a copy of the License at:
  #
  #       https://www.apache.org/licenses/LICENSE-2.0
  #
  #  Unless required by applicable law or agreed to in writing, software
  #  distributed under the License is distributed on an "AS IS" BASIS,
  #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  #  or implied. See the License for the specific language governing
  #  permissions and limitations under the License.
  """Utility functions for the CLI."""
  
  import contextlib
  import csv
  import io
  import json
  import os
  import platform
  import re
  import shutil
  import subprocess
  import sys
  from functools import partial
  from typing import (
      IO,
      TYPE_CHECKING,
      AbstractSet,
      Any,
      Callable,
      Dict,
      Iterator,
      List,
      Literal,
      NoReturn,
      Optional,
      Sequence,
      Set,
      Tuple,
      Type,
      TypeVar,
      Union,
      cast,
      get_args,
  )
  from uuid import UUID
  
  import click
  import yaml
  from pydantic import BaseModel, SecretStr
  from rich import box, table
  from rich.console import Console
  from rich.emoji import Emoji, NoEmoji
  from rich.markdown import Markdown
  from rich.padding import Padding
  from rich.prompt import Confirm, Prompt
  from rich.style import Style
  from rich.syntax import Syntax
  from rich.table import Table
  
  from zenml.client import Client
  from zenml.console import console, zenml_style_defaults
  from zenml.constants import (
      ENV_ZENML_CLI_COLUMN_WIDTH,
      ENV_ZENML_DEFAULT_OUTPUT,
      FILTERING_DATETIME_FORMAT,
      IS_DEBUG_ENV,
      handle_int_env_var,
  )
  from zenml.deployers.utils import (
      get_deployment_input_schema,
      get_deployment_invocation_example,
      get_deployment_output_schema,
  )
  from zenml.enums import (
      DeploymentStatus,
      GenericFilterOps,
      ServiceState,
      StackComponentType,
  )
  from zenml.logger import get_logger
  from zenml.model_registries.base_model_registry import (
      RegisteredModel,
      RegistryModelVersion,
  )
  from zenml.models import (
      BaseFilter,
      BaseIdentifiedResponse,
      BoolFilter,
      NumericFilter,
      Page,
      ServiceConnectorRequirements,
      StrFilter,
      UserResponse,
      UserScopedResponse,
      UUIDFilter,
  )
  from zenml.models.v2.base.filter import FilterGenerator
  from zenml.services import BaseService
  from zenml.stack.flavor import Flavor
  from zenml.stack.stack_component import StackComponentConfig
  from zenml.utils import secret_utils
  from zenml.utils.package_utils import requirement_installed
  from zenml.utils.time_utils import expires_in
  from zenml.utils.typing_utils import get_origin, is_union
  from zenml.utils.uuid_utils import is_valid_uuid
  
  if TYPE_CHECKING:
      from uuid import UUID
  
      from rich.text import Text
  
      from zenml.enums import ExecutionStatus
      from zenml.integrations.integration import Integration
      from zenml.model_deployers import BaseModelDeployer
      from zenml.models import (
          AuthenticationMethodModel,
          ComponentResponse,
          DeploymentResponse,
          FlavorResponse,
          PipelineRunResponse,
          PipelineSnapshotResponse,
          ProjectResponse,
          ResourceTypeModel,
          ServiceConnectorRequest,
          ServiceConnectorResourcesModel,
          ServiceConnectorResponse,
          ServiceConnectorTypeModel,
          StackResponse,
          UserResponse,
      )
      from zenml.stack import Stack
  
  
  logger = get_logger(__name__)
  
  AnyResponse = TypeVar(
      "AnyResponse", bound=BaseIdentifiedResponse[Any, Any, Any]
  )
  OutputFormat = Literal["table", "json", "yaml", "csv", "tsv"]
  
  MAX_ARGUMENT_VALUE_SIZE = 10240
  
  
  T = TypeVar("T", bound=BaseIdentifiedResponse)  # type: ignore[type-arg]
  
  
  def title(text: str) -> None:
      """Echo a title formatted string on the CLI.
  
      Args:
          text: Input text string.
      """
      console.print(text.upper(), style=zenml_style_defaults["title"])
  
  
  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)
  
  
  def declare(
      text: Union[str, "Text"],
      bold: Optional[bool] = None,
      italic: Optional[bool] = None,
      **kwargs: Any,
  ) -> None:
      """Echo a declaration on the CLI.
  
      Args:
          text: Input text string.
          bold: Optional boolean to bold the text.
          italic: Optional boolean to italicize the text.
          **kwargs: Optional kwargs to be passed to console.print().
      """
      base_style = zenml_style_defaults["info"]
      style = Style.chain(base_style, Style(bold=bold, italic=italic))
      console.print(text, style=style, **kwargs)
  
  
  def error(text: str) -> NoReturn:
      """Echo an error string on the CLI.
  
      Args:
          text: Input text string.
  
      Raises:
          StyledClickException: when called.
      """
      error_prefix = click.style("Error: ", fg="red", bold=True)
      error_message = click.style(text, fg="red", bold=False)
  
      class StyledClickException(click.ClickException):
          def show(self, file: Optional[IO[Any]] = None) -> None:
              if file is None:
                  file = click.get_text_stream("stderr")
              click.echo(self.message, file=file)
  
      raise StyledClickException(message=error_prefix + error_message)
  
  
  def exception(exception: Exception) -> NoReturn:
      """Echo an exception string on the CLI.
  
      Args:
          exception: Input exception.
      """
      # KeyError add quotes around their error message to handle the case when
      # someone tried to fetch something with the key '' (empty string). In all
      # other cases where the key is not empty however, this adds unnecessary
      # quotes around the error message, which this function removes.
      if (
          isinstance(exception, KeyError)
          and len(exception.args) == 1
          # If the exception is a KeyError with an empty string as the argument,
          # we use the default string representation which will print ''
          and exception.args[0]
          and isinstance(exception.args[0], str)
      ):
          error_message = exception.args[0]
      else:
          error_message = str(exception)
  
      error(error_message)
  
  
  def warning(
      text: str,
      bold: Optional[bool] = None,
      italic: Optional[bool] = None,
      **kwargs: Any,
  ) -> None:
      """Echo a warning string on the CLI.
  
      Args:
          text: Input text string.
          bold: Optional boolean to bold the text.
          italic: Optional boolean to italicize the text.
          **kwargs: Optional kwargs to be passed to console.print().
      """
      base_style = zenml_style_defaults["warning"]
      style = Style.chain(base_style, Style(bold=bold, italic=italic))
      console.print(text, style=style, **kwargs)
  
  
  def success(
      text: str,
      bold: Optional[bool] = None,
      italic: Optional[bool] = None,
      **kwargs: Any,
  ) -> None:
      """Echo a success string on the CLI.
  
      Args:
          text: Input text string.
          bold: Optional boolean to bold the text.
          italic: Optional boolean to italicize the text.
          **kwargs: Optional kwargs to be passed to console.print().
      """
      base_style = zenml_style_defaults["success"]
      style = Style.chain(base_style, Style(bold=bold, italic=italic))
      console.print(text, style=style, **kwargs)
  
  
  def print_table(
      obj: List[Dict[str, Any]],
      title: Optional[str] = None,
      caption: Optional[str] = None,
      columns: Optional[Dict[str, Union[table.Column, str]]] = None,
      column_aliases: Optional[Dict[str, str]] = None,
      **kwargs: table.Column,
  ) -> None:
      """Prints a list of dicts as a table.
  
      Args:
          obj: A List containing dictionaries.
          title: Title of the table.
          caption: Caption of the table.
          columns: Optional mapping of data keys to column configurations.
              Values can be either a rich Column object (uses .header for display)
              or a string (used directly as the header). Keys not in this mapping
              will use the uppercased key as the header.
          column_aliases: Optional mapping of original column names to display
              names. Use this to rename columns in the table output.
          **kwargs: Deprecated. Use `columns` dict instead. Kept for backward
              compatibility.
      """
      all_columns: Dict[str, Union[table.Column, str]] = dict(kwargs)
      if columns:
          all_columns.update(columns)
  
      data = obj
      if all_columns:
          data = []
          for row in obj:
              new_row = {}
              for k, v in row.items():
                  col_config = all_columns.get(k)
                  if col_config is None:
                      header = k.upper()
                  elif isinstance(col_config, table.Column):
                      header = str(col_config.header)
                  else:
                      header = str(col_config)
                  new_row[header] = v
              data.append(new_row)
  
      if not data:
          return
  
      print(
          _render_table(
              data, title=title, caption=caption, column_aliases=column_aliases
          ),
          end="",
      )
  
  
  def print_pydantic_models(
      models: Union[Page[T], List[T]],
      columns: Optional[List[str]] = None,
      exclude_columns: Optional[List[str]] = None,
      active_models: Optional[List[T]] = None,
      show_active: bool = False,
      column_aliases: Optional[Dict[str, str]] = None,
  ) -> 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.
          columns: Optionally specify subset and order of columns to display.
          exclude_columns: Optionally specify columns to exclude. (Note: `columns`
              takes precedence over `exclude_columns`.)
          active_models: Optional list of active models of the given type T.
          show_active: Flag to decide whether to append the active model on the
              top of the list.
          column_aliases: Optional mapping of original column names to display
              names. Use this to rename columns in the table output.
      """
      if exclude_columns is None:
          exclude_columns = list()
  
      show_active_column = True
      if active_models is None:
          show_active_column = False
          active_models = list()
  
      def __dictify(model: T) -> Dict[str, str]:
          """Helper function to map over the list to turn Models into dicts.
  
          Args:
              model: Pydantic model.
  
          Returns:
              Dict of model attributes.
          """
          include_columns = _extract_model_columns(
              model=model, columns=columns, exclude_columns=exclude_columns
          )
  
          items: Dict[str, Any] = {}
  
          for k in include_columns:
              value = getattr(model, k)
              items[k] = _format_response_value(value)
  
          if not active_models and not show_active:
              return items
  
          marker = "active"
          if marker in items:
              marker = "current"
          if active_models is not None and show_active_column:
              return {
                  marker: (
                      "[green]●[/green]"
                      if any(model.id == a.id for a in active_models)
                      else "-"
                  ),
                  **items,
              }
  
          return items
  
      active_ids = [a.id for a in active_models]
      if isinstance(models, Page):
          table_items = list(models.items)
  
          if show_active:
              for active_model in active_models:
                  if active_model.id not in [i.id for i in table_items]:
                      table_items.append(active_model)
  
              table_items = [i for i in table_items if i.id in active_ids] + [
                  i for i in table_items if i.id not in active_ids
              ]
  
          print_table(
              [__dictify(model) for model in table_items],
              column_aliases=column_aliases,
          )
          print_page_info(models)
      else:
          table_items = list(models)
  
          if show_active:
              for active_model in active_models:
                  if active_model.id not in [i.id for i in table_items]:
                      table_items.append(active_model)
  
              table_items = [i for i in table_items if i.id in active_ids] + [
                  i for i in table_items if i.id not in active_ids
              ]
  
          print_table(
              [__dictify(model) for model in table_items],
              column_aliases=column_aliases,
          )
  
  
  def print_pydantic_model(
      title: str,
      model: BaseModel,
      exclude_columns: Optional[AbstractSet[str]] = None,
      columns: Optional[AbstractSet[str]] = None,
  ) -> None:
      """Prints a single Pydantic model in a table.
  
      Args:
          title: Title of the table.
          model: Pydantic model that will be represented as a row in the table.
          exclude_columns: Optionally specify columns to exclude.
          columns: Optionally specify subset and order of columns to display.
      """
      include_columns = _extract_model_columns(
          model=model,
          columns=list(columns) if columns else None,
          exclude_columns=list(exclude_columns) if exclude_columns else None,
      )
  
      items: Dict[str, Any] = {}
      for k in include_columns:
          value = getattr(model, k)
          items[k] = _format_response_value(value)
  
      _print_key_value_table(items, title=title)
  
  
  def format_integration_list(
      integrations: List[Tuple[str, Type["Integration"]]],
  ) -> 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.
  
      Args:
          integrations: List of tuples containing the name of the integration and
              the integration metadata.
  
      Returns:
          List of Dicts containing the name of the integration and the integration
      """
      list_of_dicts = []
      for name, integration_impl in integrations:
          is_installed = integration_impl.check_installation()
          list_of_dicts.append(
              {
                  "INSTALLED": ":white_check_mark:" if is_installed else ":x:",
                  "INTEGRATION": name,
                  "REQUIRED_PACKAGES": ", ".join(
                      integration_impl.get_requirements()
                  ),
              }
          )
      return list_of_dicts
  
  
  def print_stack_configuration(stack: "StackResponse", active: bool) -> None:
      """Prints the configuration options of a stack.
  
      Args:
          stack: Instance of a stack model.
          active: Whether the stack is active.
      """
      stack_caption = f"'{stack.name}' stack"
      if active:
          stack_caption += " (ACTIVE)"
      rich_table = table.Table(
          box=box.ROUNDED,
          title="Stack Configuration",
          caption=stack_caption,
          show_lines=True,
      )
      rich_table.add_column("COMPONENT_TYPE", overflow="fold")
      rich_table.add_column("COMPONENT_NAME", overflow="fold")
      for component_type, components in stack.components.items():
          rich_table.add_row(component_type, components[0].name)
  
      rich_table.columns[0]._cells = [
          component.upper()  # type: ignore[union-attr]
          for component in rich_table.columns[0]._cells
      ]
      console.print(rich_table)
  
      if not stack.labels:
          declare("No labels are set for this stack.")
      else:
          rich_table = table.Table(
              box=box.ROUNDED,
              title="Labels",
              show_lines=True,
          )
          rich_table.add_column("LABEL")
          rich_table.add_column("VALUE", overflow="fold")
  
          for label, value in stack.labels.items():
              rich_table.add_row(label, str(value))
  
          console.print(rich_table)
  
      declare(
          f"Stack '{stack.name}' with id '{stack.id}' is "
          f"{f'owned by user {stack.user.name}.' if stack.user else 'unowned.'}"
      )
  
  
  def print_flavor_list(flavors: Page["FlavorResponse"]) -> None:
      """Prints the list of flavors.
  
      Args:
          flavors: List of flavors to print.
      """
      flavor_table = []
      for f in flavors.items:
          flavor_table.append(
              {
                  "FLAVOR": f.name,
                  "INTEGRATION": f.integration,
                  "SOURCE": f.source,
                  "CONNECTOR TYPE": f.connector_type or "",
                  "RESOURCE TYPE": f.connector_resource_type or "",
              }
          )
  
      print_table(flavor_table)
  
  
  def print_stack_component_configuration(
      component: "ComponentResponse",
      active_status: bool,
      connector_requirements: Optional[ServiceConnectorRequirements] = None,
  ) -> None:
      """Prints the configuration options of a stack component.
  
      Args:
          component: The stack component to print.
          active_status: Whether the stack component is active.
          connector_requirements: Connector requirements for the component, taken
              from the component flavor. Only needed if the component has a
              connector.
      """
      user_name = _get_user_name(component.user)
  
      declare(
          f"{component.type.value.title()} '{component.name}' of flavor "
          f"'{component.flavor_name}' with id '{component.id}' is owned by "
          f"user '{user_name}'."
      )
  
      if len(component.configuration) == 0:
          declare("No configuration options are set for this component.")
  
      else:
          title_ = (
              f"'{component.name}' {component.type.value.upper()} "
              f"Component Configuration"
          )
  
          if active_status:
              title_ += " (ACTIVE)"
          rich_table = table.Table(
              box=box.ROUNDED,
              title=title_,
              show_lines=True,
          )
          rich_table.add_column("COMPONENT_PROPERTY")
          rich_table.add_column("VALUE", overflow="fold")
  
          items = component.configuration.items()
          for item in items:
              elements = []
              for idx, elem in enumerate(item):
                  if idx == 0:
                      elements.append(f"{elem.upper()}")
                  else:
                      elements.append(str(elem))
              rich_table.add_row(*elements)
  
          console.print(rich_table)
  
      if not component.labels:
          declare("No labels are set for this component.")
      else:
          rich_table = table.Table(
              box=box.ROUNDED,
              title="Labels",
              show_lines=True,
          )
          rich_table.add_column("LABEL")
          rich_table.add_column("VALUE", overflow="fold")
  
          for label, value in component.labels.items():
              rich_table.add_row(label, str(value))
  
          console.print(rich_table)
  
      if not component.connector:
          declare("No connector is set for this component.")
      else:
          rich_table = table.Table(
              box=box.ROUNDED,
              title="Service Connector",
              show_lines=True,
          )
          rich_table.add_column("PROPERTY")
          rich_table.add_column("VALUE", overflow="fold")
  
          resource_type = (
              connector_requirements.resource_type
              if connector_requirements
              else component.connector.resource_types[0]
          )
  
          connector_dict = {
              "ID": str(component.connector.id),
              "NAME": component.connector.name,
              "TYPE": component.connector.type,
              "RESOURCE TYPE": resource_type,
              "RESOURCE NAME": component.connector_resource_id
              or component.connector.resource_id
              or "N/A",
          }
  
          for label, value in connector_dict.items():
              rich_table.add_row(label, value)
  
          console.print(rich_table)
  
  
  def expand_argument_value_from_file(name: str, value: str) -> str:
      """Expands the value of an argument pointing to a file into the contents of that file.
  
      Args:
          name: Name of the argument. Used solely for logging purposes.
          value: The value of the argument. This is to be interpreted as a
              filename if it begins with a `@` character.
  
      Returns:
          The argument value expanded into the contents of the file, if the
          argument value begins with a `@` character. Otherwise, the argument
          value is returned unchanged.
  
      Raises:
          ValueError: If the argument value points to a file that doesn't exist,
              that cannot be read, or is too long(i.e. exceeds
              `MAX_ARGUMENT_VALUE_SIZE` bytes).
      """
      if value.startswith("@@"):
          return value[1:]
      if not value.startswith("@"):
          return value
      filename = os.path.abspath(os.path.expanduser(value[1:]))
      logger.info(
          f"Expanding argument value `{name}` to contents of file `{filename}`."
      )
      if not os.path.isfile(filename):
          raise ValueError(
              f"Could not load argument '{name}' value: file "
              f"'{filename}' does not exist or is not readable."
          )
      try:
          if os.path.getsize(filename) > MAX_ARGUMENT_VALUE_SIZE:
              raise ValueError(
                  f"Could not load argument '{name}' value: file "
                  f"'{filename}' is too large (max size is "
                  f"{MAX_ARGUMENT_VALUE_SIZE} bytes)."
              )
  
          with open(filename, "r") as f:
              return f.read()
      except OSError as e:
          raise ValueError(
              f"Could not load argument '{name}' value: file "
              f"'{filename}' could not be accessed: {str(e)}"
          )
  
  
  def convert_structured_str_to_dict(string: str) -> Dict[str, str]:
      """Convert a structured string (JSON or YAML) into a dict.
  
      Examples:
          >>> convert_structured_str_to_dict('{"location": "Nevada", "aliens":"many"}')
          {'location': 'Nevada', 'aliens': 'many'}
          >>> convert_structured_str_to_dict('location: Nevada \\naliens: many')
          {'location': 'Nevada', 'aliens': 'many'}
          >>> convert_structured_str_to_dict("{'location': 'Nevada', 'aliens': 'many'}")
          {'location': 'Nevada', 'aliens': 'many'}
  
      Args:
          string: JSON or YAML string value
  
      Returns:
          dict_: dict from structured JSON or YAML str
      """
      try:
          dict_: Dict[str, str] = json.loads(string)
          return dict_
      except ValueError:
          pass
  
      try:
          # Here, Dict type in str is implicitly supported by yaml.safe_load()
          dict_ = yaml.safe_load(string)
          return dict_
      except yaml.YAMLError:
          pass
  
      error(
          f"Invalid argument: '{string}'. Please provide the value in JSON or YAML format."
      )
  
  
  def parse_name_and_extra_arguments(
      args: List[str],
      expand_args: bool = False,
      name_mandatory: bool = True,
  ) -> Tuple[Optional[str], Dict[str, str]]:
      """Parse a name and extra arguments from the CLI.
  
      This is a utility function used to parse a variable list of optional CLI
      arguments of the form `--key=value` that must also include one mandatory
      free-form name argument. There is no restriction as to the order of the
      arguments.
  
      Examples:
          >>> parse_name_and_extra_arguments(['foo']])
          ('foo', {})
          >>> parse_name_and_extra_arguments(['foo', '--bar=1'])
          ('foo', {'bar': '1'})
          >>> parse_name_and_extra_arguments(['--bar=1', 'foo', '--baz=2'])
          ('foo', {'bar': '1', 'baz': '2'})
          >>> parse_name_and_extra_arguments(['--bar=1'])
          Traceback (most recent call last):
              ...
              ValueError: Missing required argument: name
  
      Args:
          args: A list of command line arguments from the CLI.
          expand_args: Whether to expand argument values into the contents of the
              files they may be pointing at using the special `@` character.
          name_mandatory: Whether the name argument is mandatory.
  
      Returns:
          The name and a dict of parsed args.
      """
      name: Optional[str] = None
      # The name was not supplied as the first argument, we have to
      # search the other arguments for the name.
      for i, arg in enumerate(args):
          if not arg:
              # Skip empty arguments.
              continue
          if arg.startswith("--"):
              continue
          name = args.pop(i)
          break
      else:
          if name_mandatory:
              error(
                  "A name must be supplied. Please see the command help for more "
                  "information."
              )
  
      message = (
          "Please provide args with a proper "
          "identifier as the key and the following structure: "
          '--custom_argument="value"'
      )
      args_dict: Dict[str, str] = {}
      for a in args:
          if not a:
              # Skip empty arguments.
              continue
          if not a.startswith("--") or "=" not in a:
              error(f"Invalid argument: '{a}'. {message}")
          key, value = a[2:].split("=", maxsplit=1)
          if not key.isidentifier():
              error(f"Invalid argument: '{a}'. {message}")
          args_dict[key] = value
  
      if expand_args:
          args_dict = {
              k: expand_argument_value_from_file(k, v)
              for k, v in args_dict.items()
          }
  
      return name, args_dict
  
  
  def validate_keys(key: str) -> None:
      """Validates key if it is a valid python string.
  
      Args:
          key: key to validate
      """
      if not key.isidentifier():
          error("Please provide args with a proper identifier as the key.")
  
  
  def prompt_configuration(
      config_schema: Dict[str, Any],
      show_secrets: bool = False,
      existing_config: Optional[Dict[str, str]] = None,
  ) -> Dict[str, str]:
      """Prompt the user for configuration values using the provided schema.
  
      Args:
          config_schema: The configuration schema.
          show_secrets: Whether to show secrets in the terminal.
          existing_config: The existing configuration values.
  
      Returns:
          The configuration values provided by the user.
      """
      is_update = False
      if existing_config is not None:
          is_update = True
      existing_config = existing_config or {}
  
      config_dict = {}
      for attr_name, attr_schema in config_schema.get("properties", {}).items():
          title = attr_schema.get("title", attr_name)
          attr_type_name = attr_type = attr_schema.get("type", "string")
          if attr_type == "array":
              attr_type_name = "list (CSV or JSON)"
          title = f"[{attr_name}] {title}"
          required = attr_name in config_schema.get("required", [])
          hidden = attr_schema.get("format", "") == "password"
          subtitles: List[str] = []
          subtitles.append(attr_type_name)
          if hidden:
              subtitles.append("secret")
          if required:
              subtitles.append("required")
          else:
              subtitles.append("optional")
          if subtitles:
              title += f" {{{', '.join(subtitles)}}}"
  
          existing_value = existing_config.get(attr_name)
          if is_update:
              if existing_value:
                  if isinstance(existing_value, SecretStr):
                      existing_value = existing_value.get_secret_value()
                  if hidden and not show_secrets:
                      title += " is currently set to: [HIDDEN]"
                  else:
                      if attr_type == "array":
                          existing_value = json.dumps(existing_value)
                      title += f" is currently set to: '{existing_value}'"
              else:
                  title += " is not currently set"
  
              click.echo(title)
  
              if existing_value:
                  title = (
                      "Please enter a new value or press Enter to keep the "
                      "existing one"
                  )
              elif required:
                  title = "Please enter a new value"
              else:
                  title = "Please enter a new value or press Enter to skip it"
  
          while True:
              # Ask the user to enter a value for the attribute
              value = click.prompt(
                  title,
                  type=str,
                  hide_input=hidden and not show_secrets,
                  default=existing_value or ("" if not required else None),
                  show_default=False,
              )
              if not value:
                  if required:
                      warning(
                          f"The attribute '{title}' is mandatory. "
                          "Please enter a non-empty value."
                      )
                      continue
                  else:
                      value = None
                      break
              else:
                  break
  
          if (
              is_update
              and value is not None
              and value == existing_value
              and not required
          ):
              confirm = click.confirm(
                  "You left this optional attribute unchanged. Would you "
                  "like to remove its value instead?",
                  default=False,
              )
              if confirm:
                  value = None
  
          if value:
              config_dict[attr_name] = value
  
      return config_dict
  
  
  def install_packages(
      packages: List[str],
      upgrade: bool = False,
      use_uv: bool = False,
  ) -> None:
      """Installs pypi packages into the current environment with pip or uv.
  
      When using with `uv`, a virtual environment is required.
  
      Args:
          packages: List of packages to install.
          upgrade: Whether to upgrade the packages if they are already installed.
          use_uv: Whether to use uv for package installation.
  
      Raises:
          e: If the package installation fails.
      """
      if "neptune" in packages:
          declare(
              "Uninstalling legacy `neptune-client` package to avoid version "
              "conflicts with new `neptune` package..."
          )
          uninstall_package("neptune-client")
  
      if "prodigy" in packages:
          packages.remove("prodigy")
          declare(
              "The `prodigy` package should be installed manually using your "
              "license key. Please visit https://prodi.gy/docs/install for more "
              "information."
          )
      if not packages:
          # if user only tried to install prodigy, we can
          # just return without doing anything
          return
  
      if use_uv and not requirement_installed("uv"):
          # If uv is installed globally, don't run as a python module
          command = []
      else:
          command = [sys.executable, "-m"]
  
      command += ["uv", "pip", "install"] if use_uv else ["pip", "install"]
  
      if upgrade:
          command += ["--upgrade"]
  
      command += packages
  
      if not IS_DEBUG_ENV:
          quiet_flag = "-q" if use_uv else "-qqq"
          command.append(quiet_flag)
          if not use_uv:
              command.append("--no-warn-conflicts")
  
      try:
          subprocess.check_call(command)
      except subprocess.CalledProcessError as e:
          if (
              use_uv
              and "Failed to locate a virtualenv or Conda environment" in str(e)
          ):
              error(
                  "Failed to locate a virtualenv or Conda environment. "
                  "When using uv, a virtual environment is required. "
                  "Run `uv venv` to create a virtualenv and retry."
              )
          else:
              raise e
  
  
  def uninstall_package(package: str, use_uv: bool = False) -> None:
      """Uninstalls pypi package from the current environment with pip or uv.
  
      Args:
          package: The package to uninstall.
          use_uv: Whether to use uv for package uninstallation.
      """
      if use_uv and not requirement_installed("uv"):
          command = []
      else:
          command = [sys.executable, "-m"]
  
      command += (
          ["uv", "pip", "uninstall", "-q"]
          if use_uv
          else ["pip", "uninstall", "-y", "-qqq"]
      )
      command += [package]
  
      subprocess.check_call(command)
  
  
  def is_uv_installed() -> bool:
      """Check if uv is installed.
  
      Returns:
          True if uv is installed, False otherwise.
      """
      return shutil.which("uv") is not None
  
  
  def is_pip_installed() -> bool:
      """Check if pip is installed in the current environment.
  
      Returns:
          True if pip is installed, False otherwise.
      """
      return requirement_installed("pip")
  
  
  def pretty_print_secret(
      secret: Dict[str, str],
      hide_secret: bool = True,
  ) -> None:
      """Print all key-value pairs associated with a secret.
  
      Args:
          secret: Secret values to print.
          hide_secret: boolean that configures if the secret values are shown
              on the CLI
      """
      title: Optional[str] = None
  
  