python/google/protobuf/internal/field_mask.py

# Protocol Buffers - Google's data interchange format
# Copyright 2008 Google Inc.  All rights reserved.
#
# Use of this source code is governed by a BSD-style
# license that can be found in the LICENSE file or at
# https://developers.google.com/open-source/licenses/bsd
"""Contains FieldMask class."""

from google.protobuf.descriptor import FieldDescriptor


class FieldMask(object):
  """Class for FieldMask message type."""

  __slots__ = ()

  def ToJsonString(self):
    """Converts FieldMask to string according to ProtoJSON spec."""
    camelcase_paths = []
    for path in self.paths:
      camelcase_paths.append(_SnakeCaseToCamelCase(path))
    return ','.join(camelcase_paths)

  def FromJsonString(self, value):
    """Converts string to FieldMask according to ProtoJSON spec."""
    if not isinstance(value, str):
      raise ValueError('FieldMask JSON value not a string: {!r}'.format(value))
    self.Clear()
    if value:
      for path in value.split(','):
        self.paths.append(_CamelCaseToSnakeCase(path))

  def IsValidForDescriptor(self, message_descriptor):
    """Checks whether the FieldMask is valid for Message Descriptor."""
    for path in self.paths:
      if not _IsValidPath(message_descriptor, path):
        return False
    return True

  def AllFieldsFromDescriptor(self, message_descriptor):
    """Gets all direct fields of Message Descriptor to FieldMask."""
    self.Clear()
    for field in message_descriptor.fields:
      self.paths.append(field.name)

  def CanonicalFormFromMask(self, mask):
    """Converts a FieldMask to the canonical form.

    Removes paths that are covered by another path. For example,
    "foo.bar" is covered by "foo" and will be removed if "foo"
    is also in the FieldMask. Then sorts all paths in alphabetical order.

    Args:
      mask: The original FieldMask to be converted.
    """
    tree = _FieldMaskTree(mask)
    tree.ToFieldMask(self)

  def Union(self, mask1, mask2):
    """Merges mask1 and mask2 into this FieldMask."""
    _CheckFieldMaskMessage(mask1)
    _CheckFieldMaskMessage(mask2)
    tree = _FieldMaskTree(mask1)
    tree.MergeFromFieldMask(mask2)
    tree.ToFieldMask(self)

  def Intersect(self, mask1, mask2):
    """Intersects mask1 and mask2 into this FieldMask."""
    _CheckFieldMaskMessage(mask1)
    _CheckFieldMaskMessage(mask2)
    tree = _FieldMaskTree(mask1)
    intersection = _FieldMaskTree()
    for path in mask2.paths:
      tree.IntersectPath(path, intersection)
    intersection.ToFieldMask(self)

  def MergeMessage(
      self,
      source,
      destination,
      replace_message_field=False,
      replace_repeated_field=False,
  ):
    """Merges fields specified in FieldMask from source to destination.

    Args:
      source: Source message.
      destination: The destination message to be merged into.
      replace_message_field: Replace message field if True. Merge message field
        if False.
      replace_repeated_field: Replace repeated field if True. Append elements of
        repeated field if False.
    """
    tree = _FieldMaskTree(self)
    tree.MergeMessage(
        source, destination, replace_message_field, replace_repeated_field
    )


def _IsValidPath(message_descriptor, path):
  """Checks whether the path is valid for Message Descriptor."""
  parts = path.split('.')
  last = parts.pop()
  for name in parts:
    field = message_descriptor.fields_by_name.get(name)
    if (
        field is None
        or field.is_repeated
        or field.type != FieldDescriptor.TYPE_MESSAGE
    ):
      return False
    message_descriptor = field.message_type
  return last in message_descriptor.fields_by_name


def _CheckFieldMaskMessage(message):
  """Raises ValueError if message is not a FieldMask."""
  message_descriptor = message.DESCRIPTOR
  if (
      message_descriptor.name != 'FieldMask'
      or message_descriptor.file.name != 'google/protobuf/field_mask.proto'
  ):
    raise ValueError(
        'Message {0} is not a FieldMask.'.format(message_descriptor.full_name)
    )


def _SnakeCaseToCamelCase(path_name):
  """Converts a path name from snake_case to camelCase."""
  result = []
  after_underscore = False
  for c in path_name:
    if c.isupper():
      raise ValueError(
          'Fail to print FieldMask to Json string: Path name '
          '{0} must not contain uppercase letters.'.format(path_name)
      )
    if after_underscore:
      if c.islower():
        result.append(c.upper())
        after_underscore = False
      else:
        raise ValueError(
            'Fail to print FieldMask to Json string: The '
            'character after a "_" must be a lowercase letter '
            'in path name {0}.'.format(path_name)
        )
    elif c == '_':
      after_underscore = True
    else:
      result += c

  if after_underscore:
    raise ValueError(
        'Fail to print FieldMask to Json string: Trailing "_" '
        'in path name {0}.'.format(path_name)
    )
  return ''.join(result)


def _CamelCaseToSnakeCase(path_name):
  """Converts a field name from camelCase to snake_case."""
  result = []
  for c in path_name:
    if c == '_':
      raise ValueError(
          'Fail to parse FieldMask: Path name '
          '{0} must not contain "_"s.'.format(path_name)
      )
    if c.isupper():
      result += '_'
      result += c.lower()
    else:
      result += c
  return ''.join(result)


class _FieldMaskTree(object):
  """Represents a FieldMask in a tree structure.

  For example, given a FieldMask "foo.bar,foo.baz,bar.baz",
  the FieldMaskTree will be:
      [_root] -+- foo -+- bar
            |       |
            |       +- baz
            |
            +- bar --- baz
  In the tree, each leaf node represents a field path.
  """

  __slots__ = ('_root',)

  def __init__(self, field_mask=None):
    """Initializes the tree by FieldMask."""
    self._root = {}
    if field_mask:
      self.MergeFromFieldMask(field_mask)

  def MergeFromFieldMask(self, field_mask):
    """Merges a FieldMask to the tree."""
    for path in field_mask.paths:
      self.AddPath(path)

  def AddPath(self, path):
    """Adds a field path into the tree.

    If the field path to add is a sub-path of an existing field path
    in the tree (i.e., a leaf node), it means the tree already matches
    the given path so nothing will be added to the tree. If the path
    matches an existing non-leaf node in the tree, that non-leaf node
    will be turned into a leaf node with all its children removed because
    the path matches all the node's children. Otherwise, a new path will
    be added.

    Args:
      path: The field path to add.
    """
    node = self._root
    for name in path.split('.'):
      if name not in node:
        node[name] = {}
      elif not node[name]:
        # Pre-existing empty node implies we already have this entire tree.
        return
      node = node[name]
    # Remove any sub-trees we might have had.
    node.clear()

  def ToFieldMask(self, field_mask):
    """Converts the tree to a FieldMask."""
    field_mask.Clear()
    _AddFieldPaths(self._root, '', field_mask)

  def IntersectPath(self, path, intersection):
    """Calculates the intersection part of a field path with this tree.

    Args:
      path: The field path to calculates.
      intersection: The out tree to record the intersection part.
    """
    node = self._root
    for name in path.split('.'):
      if name not in node:
        return
      elif not node[name]:
        intersection.AddPath(path)
        return
      node = node[name]
    intersection.AddLeafNodes(path, node)

  def AddLeafNodes(self, prefix, node):
    """Adds leaf nodes begin with prefix to this tree."""
    if not node:
      self.AddPath(prefix)
      return
    stack = [(prefix, node)]
    while stack:
      current_prefix, current_node = stack.pop()
      if not current_node:
        self.AddPath(current_prefix)
        continue
      for name in current_node:
        child_path = current_prefix + '.' + name
        stack.append((child_path, current_node[name]))

  def MergeMessage(
      self, source, destination, replace_message, replace_repeated
  ):
    """Merge all fields specified by this tree from source to destination."""
    _MergeMessage(
        self._root, source, destination, replace_message, replace_repeated
    )


def _StrConvert(value):
  """Converts value to str if it is not."""
  # This file is imported by c extension and some methods like ClearField
  # requires string for the field name. py2/py3 has different text
  # type and may use unicode.
  if not isinstance(value, str):
    return value.encode('utf-8')
  return value


def _MergeMessage(node, source, destination, replace_message, replace_repeated):
  """Merge all fields specified by a sub-tree from source to destination."""
  stack = [(node, source, destination)]
  while stack:
    current_node, current_source, current_destination = stack.pop()
    source_descriptor = current_source.DESCRIPTOR
    for name in current_node:
      child = current_node[name]
      field = source_descriptor.fields_by_name[name]
      if field is None:
        raise ValueError(
            "Error: Can't find field {0} in message {1}.".format(
                name, source_descriptor.full_name
            )
        )
      if child:
        # Sub-paths are only allowed for singular message fields.
        if (
            field.is_repeated
            or field.cpp_type != FieldDescriptor.CPPTYPE_MESSAGE
        ):
          raise ValueError(
              'Error: Field {0} in message {1} is not a singular '
              'message field and cannot have sub-fields.'.format(
                  name, source_descriptor.full_name
              )
          )
        if current_source.HasField(name):
          stack.append((
              child,
              getattr(current_source, name),
              getattr(current_destination, name),
          ))
        continue
      if field.is_repeated:
        if replace_repeated:
          current_destination.ClearField(_StrConvert(name))
        repeated_source = getattr(current_source, name)
        repeated_destination = getattr(current_destination, name)
        repeated_destination.MergeFrom(repeated_source)
      else:
        if field.cpp_type == FieldDescriptor.CPPTYPE_MESSAGE:
          if replace_message:
            current_destination.ClearField(_StrConvert(name))
          if current_source.HasField(name):
            getattr(current_destination, name).MergeFrom(
                getattr(current_source, name)
            )
        elif not field.has_presence or current_source.HasField(name):
          setattr(current_destination, name, getattr(current_source, name))
        else:
          current_destination.ClearField(_StrConvert(name))


def _AddFieldPaths(node, prefix, field_mask):
  """Adds the field paths descended from node to field_mask."""
  stack = [(node, prefix)]
  while stack:
    current_node, current_prefix = stack.pop()
    if not current_node and current_prefix:
      field_mask.paths.append(current_prefix)
      continue
    for name in sorted(current_node, reverse=True):
      if current_prefix:
        child_path = current_prefix + '.' + name
      else:
        child_path = name
      stack.append((current_node[name], child_path))