Source code for proto.fields

# Copyright 2018 Google LLC
# 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

from google.protobuf import descriptor_pb2

from proto.primitives import ProtoType

[docs]class Field: """A representation of a type of field in protocol buffers.""" def __init__(self, proto_type, *, number: int, message=None, enum=None, oneof: str = None, json_name: str = None): # This class is not intended to stand entirely alone; # data is augmented by the metaclass for Message. self.mcls_data = {} self.parent = None # If the proto type sent is an object or a string, it is really # a message or enum. if not isinstance(proto_type, int): message = proto_type proto_type = ProtoType.MESSAGE # Save the direct arguments. self.number = number self.proto_type = proto_type self.message = message self.enum = enum self.oneof = oneof self.json_name = json_name # Fields are neither repeated nor maps. # The RepeatedField and MapField subclasses override these values # in their initializers. self.repeated = False # Once the descriptor is accessed the first time, cache it. # This is important because in rare cases the message or enum # types are written later. self._descriptor = None @property def descriptor(self): """Return the descriptor for the field.""" if not self._descriptor: # Resolve the message type, if any, to a string. type_name = None if isinstance(self.message, str): if not self.message.startswith(self.package): self.message = '{package}.{name}'.format( package=self.package, name=self.message, ) type_name = self.message elif self.message: if hasattr(self.message, 'DESCRIPTOR'): type_name = self.message.DESCRIPTOR.full_name else: type_name = self.message.meta.full_name elif self.enum: # FIXME: This is obviously wrong (however, it does *work*). # We need to set the enum type name and add the enum to # the descriptor (like with messages above). self.proto_type = ProtoType.INT32 # Set the descriptor. self._descriptor = descriptor_pb2.FieldDescriptorProto(, number=self.number, label=3 if self.repeated else 1, type=self.proto_type, type_name=type_name, json_name=self.json_name, ) # Return the descriptor. return self._descriptor @property def name(self) -> str: """Return the name of the field.""" return self.mcls_data['name'] @property def package(self) -> str: """Return the package of the field.""" return self.mcls_data['package'] @property def pb_type(self): """Return the composite type of the field, or None for primitives.""" # For primitives, return None. if not self.message: return None # Return the internal protobuf message. if hasattr(self.message, '_meta'): return self.message.pb() return self.message
[docs]class RepeatedField(Field): """A representation of a repeated field in protocol buffers.""" def __init__(self, proto_type, *, number: int, message=None, enum=None): super().__init__(proto_type, number=number, message=message, enum=enum) self.repeated = True
[docs]class MapField(Field): """A representation of a map field in protocol buffers.""" def __init__(self, key_type, value_type, *, number: int, message=None, enum=None): super().__init__(value_type, number=number, message=message, enum=enum) self.map_key_type = key_type
__all__ = ( 'Field', 'MapField', 'RepeatedField', )