Spaces:
Running
Running
| # 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 proto3 JSON 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 proto3 JSON 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.label == FieldDescriptor.LABEL_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) | |
| for name in node: | |
| child_path = prefix + '.' + name | |
| self.AddLeafNodes(child_path, 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.""" | |
| source_descriptor = source.DESCRIPTOR | |
| for name in node: | |
| child = 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.label == FieldDescriptor.LABEL_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 source.HasField(name): | |
| _MergeMessage( | |
| child, getattr(source, name), getattr(destination, name), | |
| replace_message, replace_repeated) | |
| continue | |
| if field.label == FieldDescriptor.LABEL_REPEATED: | |
| if replace_repeated: | |
| destination.ClearField(_StrConvert(name)) | |
| repeated_source = getattr(source, name) | |
| repeated_destination = getattr(destination, name) | |
| repeated_destination.MergeFrom(repeated_source) | |
| else: | |
| if field.cpp_type == FieldDescriptor.CPPTYPE_MESSAGE: | |
| if replace_message: | |
| destination.ClearField(_StrConvert(name)) | |
| if source.HasField(name): | |
| getattr(destination, name).MergeFrom(getattr(source, name)) | |
| else: | |
| setattr(destination, name, getattr(source, name)) | |
| def _AddFieldPaths(node, prefix, field_mask): | |
| """Adds the field paths descended from node to field_mask.""" | |
| if not node and prefix: | |
| field_mask.paths.append(prefix) | |
| return | |
| for name in sorted(node): | |
| if prefix: | |
| child_path = prefix + '.' + name | |
| else: | |
| child_path = name | |
| _AddFieldPaths(node[name], child_path, field_mask) | |