Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/proto/_file_info.py: 94%

1# Copyright 2018 Google LLC 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# https://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15import collections 

16import inspect 

17import logging 

18 

19from google.protobuf import descriptor_pb2, descriptor_pool, message, reflection 

20 

21from proto.marshal.rules.message import MessageRule 

22 

23log = logging.getLogger("_FileInfo") 

24 

25 

26class _FileInfo( 

27 collections.namedtuple( 

28 "_FileInfo", 

29 ["descriptor", "messages", "enums", "name", "nested", "nested_enum"], 

30 ) 

31): 

32 registry = {} # Mapping[str, '_FileInfo'] 

33 

34 @classmethod 

35 def maybe_add_descriptor(cls, filename, package): 

36 descriptor = cls.registry.get(filename) 

37 if not descriptor: 

38 descriptor = cls.registry[filename] = cls( 

39 descriptor=descriptor_pb2.FileDescriptorProto( 

40 name=filename, 

41 package=package, 

42 syntax="proto3", 

43 ), 

44 enums=collections.OrderedDict(), 

45 messages=collections.OrderedDict(), 

46 name=filename, 

47 nested={}, 

48 nested_enum={}, 

49 ) 

50 

51 return descriptor 

52 

53 @staticmethod 

54 def proto_file_name(name): 

55 return "{0}.proto".format(name.replace(".", "/")) 

56 

57 def _get_manifest(self, new_class): 

58 module = inspect.getmodule(new_class) 

59 if hasattr(module, "__protobuf__"): 

60 return frozenset(module.__protobuf__.manifest) 

61 

62 return frozenset() 

63 

64 def _get_remaining_manifest(self, new_class): 

65 return self._get_manifest(new_class) - {new_class.__name__} 

66 

67 def _calculate_salt(self, new_class, fallback): 

68 manifest = self._get_manifest(new_class) 

69 if manifest and new_class.__name__ not in manifest: 

70 log.warning( 

71 "proto-plus module {module} has a declared manifest but {class_name} is not in it".format( 

72 module=inspect.getmodule(new_class).__name__, 

73 class_name=new_class.__name__, 

74 ) 

75 ) 

76 

77 return "" if new_class.__name__ in manifest else (fallback or "").lower() 

78 

79 def generate_file_pb(self, new_class, fallback_salt=""): 

80 """Generate the descriptors for all protos in the file. 

81 

82 This method takes the file descriptor attached to the parent 

83 message and generates the immutable descriptors for all of the 

84 messages in the file descriptor. (This must be done in one fell 

85 swoop for immutability and to resolve proto cross-referencing.) 

86 

87 This is run automatically when the last proto in the file is 

88 generated, as determined by the module's __all__ tuple. 

89 """ 

90 pool = descriptor_pool.Default() 

91 

92 # Salt the filename in the descriptor. 

93 # This allows re-use of the filename by other proto messages if 

94 # needed (e.g. if __all__ is not used). 

95 salt = self._calculate_salt(new_class, fallback_salt) 

96 self.descriptor.name = "{name}.proto".format( 

97 name="_".join([self.descriptor.name[:-6], salt]).rstrip("_"), 

98 ) 

99 

100 # Add the file descriptor. 

101 pool.Add(self.descriptor) 

102 

103 # Adding the file descriptor to the pool created a descriptor for 

104 # each message; go back through our wrapper messages and associate 

105 # them with the internal protobuf version. 

106 for full_name, proto_plus_message in self.messages.items(): 

107 # Get the descriptor from the pool, and create the protobuf 

108 # message based on it. 

109 descriptor = pool.FindMessageTypeByName(full_name) 

110 pb_message = reflection.GeneratedProtocolMessageType( 

111 descriptor.name, 

112 (message.Message,), 

113 {"DESCRIPTOR": descriptor, "__module__": None}, 

114 ) 

115 

116 # Register the message with the marshal so it is wrapped 

117 # appropriately. 

118 # 

119 # We do this here (rather than at class creation) because it 

120 # is not until this point that we have an actual protobuf 

121 # message subclass, which is what we need to use. 

122 proto_plus_message._meta._pb = pb_message 

123 proto_plus_message._meta.marshal.register( 

124 pb_message, MessageRule(pb_message, proto_plus_message) 

125 ) 

126 

127 # Iterate over any fields on the message and, if their type 

128 # is a message still referenced as a string, resolve the reference. 

129 for field in proto_plus_message._meta.fields.values(): 

130 if field.message and isinstance(field.message, str): 

131 field.message = self.messages[field.message] 

132 elif field.enum and isinstance(field.enum, str): 

133 field.enum = self.enums[field.enum] 

134 

135 # Same thing for enums 

136 for full_name, proto_plus_enum in self.enums.items(): 

137 descriptor = pool.FindEnumTypeByName(full_name) 

138 proto_plus_enum._meta.pb = descriptor 

139 

140 # We no longer need to track this file's info; remove it from 

141 # the module's registry and from this object. 

142 self.registry.pop(self.name) 

143 

144 def ready(self, new_class): 

145 """Return True if a file descriptor may added, False otherwise. 

146 

147 This determine if all the messages that we plan to create have been 

148 created, as best as we are able. 

149 

150 Since messages depend on one another, we create descriptor protos 

151 (which reference each other using strings) and wait until we have 

152 built everything that is going to be in the module, and then 

153 use the descriptor protos to instantiate the actual descriptors in 

154 one fell swoop. 

155 

156 Args: 

157 new_class (~.MessageMeta): The new class currently undergoing 

158 creation. 

159 """ 

160 # If there are any nested descriptors that have not been assigned to 

161 # the descriptors that should contain them, then we are not ready. 

162 if len(self.nested) or len(self.nested_enum): 

163 return False 

164 

165 # If there are any unresolved fields (fields with a composite message 

166 # declared as a string), ensure that the corresponding message is 

167 # declared. 

168 for field in self.unresolved_fields: 

169 if (field.message and field.message not in self.messages) or ( 

170 field.enum and field.enum not in self.enums 

171 ): 

172 return False 

173 

174 # If the module in which this class is defined provides a 

175 # __protobuf__ property, it may have a manifest. 

176 # 

177 # Do not generate the file descriptor until every member of the 

178 # manifest has been populated. 

179 module = inspect.getmodule(new_class) 

180 manifest = self._get_remaining_manifest(new_class) 

181 

182 # We are ready if all members have been populated. 

183 return all(hasattr(module, i) for i in manifest) 

184 

185 @property 

186 def unresolved_fields(self): 

187 """Return fields with referencing message types as strings.""" 

188 for proto_plus_message in self.messages.values(): 

189 for field in proto_plus_message._meta.fields.values(): 

190 if (field.message and isinstance(field.message, str)) or ( 

191 field.enum and isinstance(field.enum, str) 

192 ): 

193 yield field