HEX
Server: Apache/2.4.65 (Ubuntu)
System: Linux ielts-store-v2 6.8.0-1036-gcp #38~22.04.1-Ubuntu SMP Thu Aug 14 01:19:18 UTC 2025 x86_64
User: root (0)
PHP: 7.2.34-54+ubuntu20.04.1+deb.sury.org+1
Disabled: pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,
Upload Files
File: //snap/google-cloud-cli/396/lib/googlecloudsdk/command_lib/run/integrations/formatters/base.py
# -*- coding: utf-8 -*- #
# Copyright 2023 Google LLC. 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
#
#    http://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.
"""Base formatter for Cloud Run Integrations."""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals

import abc

from typing import Optional
from googlecloudsdk.api_lib.run.integrations import types_utils
from googlecloudsdk.command_lib.run.integrations.formatters import states
from googlecloudsdk.core import properties
from googlecloudsdk.core.console import console_attr
from googlecloudsdk.core.resource import custom_printer_base as cp
from googlecloudsdk.generated_clients.apis.runapps.v1alpha1 import runapps_v1alpha1_messages as runapps


# Constants used to pick the symbol displayed to the console.
SUCCESS = 'SUCCESS'
UPDATING = 'UPDATING'
FAILED = 'FAILED'
MISSING = 'MISSING'
DEFAULT = 'DEFAULT'

# Constants used to pick the encoding
ASCII = 'ascii'
UTF8 = 'utf8'


class Record(object):
  """Record holds data that is passed around to printers for formatting.

  Attributes:
    name: str, name of the integration
    region: str, GCP region for the integration.
    metadata: the type metadata for the integration.
    resource: the resource of the integration.
    status: dict, application status for the given integration.
    latest_deployment: str, canonical deployment name for the latest deployment
      for the given integration.
  """

  def __init__(
      self,
      name: Optional[str],
      region: Optional[str],
      metadata: Optional[types_utils.TypeMetadata] = None,
      resource: Optional[runapps.Resource] = None,
      status: Optional[runapps.ResourceStatus] = None,
      latest_deployment: runapps.Deployment = None,
  ):
    self.name = name
    self.region = region
    self.metadata = metadata
    self.resource = resource if resource else runapps.Resource()
    self.status = status if status else runapps.ResourceStatus()
    self.latest_deployment = latest_deployment


class BaseFormatter(abc.ABC):
  """Prints the run Integration in a custom human-readable format."""

  @abc.abstractmethod
  def TransformConfig(self, record: Record) -> cp._Marker:
    """Override to describe the format of the config of the integration."""

  @abc.abstractmethod
  def TransformComponentStatus(self, record: Record) -> cp._Marker:
    """Override to describe the format of the components and status of the integration."""

  def CallToAction(self, record: Record) -> Optional[str]:
    """Override to return call to action message.

    Args:
      record: dict, the integration.

    Returns:
      A formatted string of the call to action message,
      or None if no call to action is required.
    """
    del record  # Unused
    return None

  def PrintType(self, ctype):
    """Return the type in a user friendly format.

    Args:
      ctype: the type name to be formatted.

    Returns:
      A formatted string.
    """
    return (ctype
            .replace('google_', '')
            .replace('compute_', '')
            .replace('_', ' ')
            .title())

  def GetResourceState(self, resource):
    """Return the state of the top level resource in the integration.

    Args:
      resource: dict, resource status of the integration resource.

    Returns:
      The state string.
    """
    return resource.get('state', states.UNKNOWN)

  def PrintStatus(self, status):
    """Print the status with symbol and color.

    Args:
      status: string, the status.

    Returns:
      The formatted string.
    """
    return '{} {}'.format(self.StatusSymbolAndColor(status), status)

  def StatusSymbolAndColor(self, status: str) -> str:
    """Return the color symbol for the status.

    Args:
      status: string, the status.

    Returns:
      The symbol string.
    """
    if status == states.DEPLOYED or status == states.ACTIVE:
      return GetSymbol(SUCCESS)
    if status in (states.PROVISIONING, states.UPDATING, states.NOT_READY):
      return GetSymbol(UPDATING)
    if status == states.MISSING:
      return GetSymbol(MISSING)
    if status == states.FAILED:
      return GetSymbol(FAILED)
    return GetSymbol(DEFAULT)


def GetSymbol(status, encoding=None) -> str:
  """Chooses a symbol to be displayed to the console based on the status.

  Args:
    status: str, defined as a constant in this file.  CloudSDK must
      support Python 2 at the moment so we cannot use the actual enum class.
      If the value is not valid or supported then it will return a default
      symbol.

    encoding: str, defined as a constant in this file.  If not provided, the
      encoding will be fetched from the user's console.

  Returns:
    Symbol (str) to be displayed to the console.
  """
  con = console_attr.GetConsoleAttr()
  if encoding is None:
    encoding = _GetEncoding()

  default_symbol = con.Colorize('~', 'blue')
  status_to_symbol = {
      SUCCESS: con.Colorize(
          _PickSymbol('\N{HEAVY CHECK MARK}', '+', encoding), 'green'
      ),
      UPDATING: con.Colorize(
          _PickSymbol('\N{HORIZONTAL ELLIPSIS}', '.', encoding), 'yellow'
      ),
      FAILED: con.Colorize('X', 'red'),
      MISSING: con.Colorize('?', 'yellow'),
      DEFAULT: default_symbol,
  }

  return status_to_symbol.get(status, default_symbol)


def _GetEncoding():
  """Returns the encoding used by the user's console.

  If the user has color disabled, then we will default to ascii.
  """
  if properties.VALUES.core.disable_color.GetBool():
    return ASCII

  return console_attr.GetConsoleAttr().GetEncoding()


def _PickSymbol(best, alt, encoding):
  """Chooses the best symbol (if it's in this encoding) or an alternate.

  Args:
    best: str, the symbol to return if the encoding allows.
    alt: str, alternative to return if best cannot be encoded.
    encoding:  str, possible values are utf8, ascii, and win.

  Returns:
    The symbol string if the encoding allows, otherwise an alternative string.
  """
  try:
    best.encode(encoding)
    return best
  except UnicodeError:
    return alt