| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| from enum import EnumMeta |
|
|
| from google.protobuf import descriptor_pb2 |
| from google.protobuf.internal.enum_type_wrapper import EnumTypeWrapper |
|
|
| from proto.primitives import ProtoType |
|
|
|
|
| class Field: |
| """A representation of a type of field in protocol buffers.""" |
|
|
| |
| |
| repeated = False |
|
|
| def __init__( |
| self, |
| proto_type, |
| *, |
| number: int, |
| message=None, |
| enum=None, |
| oneof: str = None, |
| json_name: str = None, |
| optional: bool = False |
| ): |
| |
| |
| self.mcls_data = None |
| self.parent = None |
|
|
| |
| |
| if not isinstance(proto_type, int): |
| |
| |
| if isinstance(proto_type, (EnumMeta, EnumTypeWrapper)): |
| enum = proto_type |
| proto_type = ProtoType.ENUM |
| else: |
| message = proto_type |
| proto_type = ProtoType.MESSAGE |
|
|
| |
| self.number = number |
| self.proto_type = proto_type |
| self.message = message |
| self.enum = enum |
| self.json_name = json_name |
| self.optional = optional |
| self.oneof = oneof |
|
|
| |
| |
| |
| self._descriptor = None |
|
|
| @property |
| def descriptor(self): |
| """Return the descriptor for the field.""" |
| if not self._descriptor: |
| |
| 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: |
| type_name = ( |
| self.message.DESCRIPTOR.full_name |
| if hasattr(self.message, "DESCRIPTOR") |
| else self.message._meta.full_name |
| ) |
| elif isinstance(self.enum, str): |
| if not self.enum.startswith(self.package): |
| self.enum = "{package}.{name}".format( |
| package=self.package, |
| name=self.enum, |
| ) |
| type_name = self.enum |
| elif self.enum: |
| type_name = ( |
| self.enum.DESCRIPTOR.full_name |
| if hasattr(self.enum, "DESCRIPTOR") |
| else self.enum._meta.full_name |
| ) |
|
|
| |
| self._descriptor = descriptor_pb2.FieldDescriptorProto( |
| name=self.name, |
| number=self.number, |
| label=3 if self.repeated else 1, |
| type=self.proto_type, |
| type_name=type_name, |
| json_name=self.json_name, |
| proto3_optional=self.optional, |
| ) |
|
|
| |
| 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 the primitive type if a primitive.""" |
| |
| if self.enum: |
| return self.enum |
|
|
| |
| |
| if not self.message: |
| return self.proto_type |
|
|
| |
| if hasattr(self.message, "_meta"): |
| return self.message.pb() |
| return self.message |
|
|
|
|
| class RepeatedField(Field): |
| """A representation of a repeated field in protocol buffers.""" |
|
|
| repeated = True |
|
|
|
|
| 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", |
| ) |
|
|