Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/protobuf/internal/extension_dict.py: 20%

1# Protocol Buffers - Google's data interchange format 

2# Copyright 2008 Google Inc. All rights reserved. 

3# 

4# Use of this source code is governed by a BSD-style 

5# license that can be found in the LICENSE file or at 

6# https://developers.google.com/open-source/licenses/bsd 

7 

8"""Contains _ExtensionDict class to represent extensions. 

9""" 

10 

11from google.protobuf.internal import type_checkers 

12from google.protobuf.descriptor import FieldDescriptor 

13 

14 

15def _VerifyExtensionHandle(message, extension_handle): 

16 """Verify that the given extension handle is valid.""" 

17 

18 if not isinstance(extension_handle, FieldDescriptor): 

19 raise KeyError('HasExtension() expects an extension handle, got: %s' % 

20 extension_handle) 

21 

22 if not extension_handle.is_extension: 

23 raise KeyError('"%s" is not an extension.' % extension_handle.full_name) 

24 

25 if not extension_handle.containing_type: 

26 raise KeyError('"%s" is missing a containing_type.' 

27 % extension_handle.full_name) 

28 

29 if extension_handle.containing_type is not message.DESCRIPTOR: 

30 raise KeyError('Extension "%s" extends message type "%s", but this ' 

31 'message is of type "%s".' % 

32 (extension_handle.full_name, 

33 extension_handle.containing_type.full_name, 

34 message.DESCRIPTOR.full_name)) 

35 

36 

37# TODO: Unify error handling of "unknown extension" crap. 

38# TODO: Support iteritems()-style iteration over all 

39# extensions with the "has" bits turned on? 

40class _ExtensionDict(object): 

41 

42 """Dict-like container for Extension fields on proto instances. 

43 

44 Note that in all cases we expect extension handles to be 

45 FieldDescriptors. 

46 """ 

47 

48 def __init__(self, extended_message): 

49 """ 

50 Args: 

51 extended_message: Message instance for which we are the Extensions dict. 

52 """ 

53 self._extended_message = extended_message 

54 

55 def __getitem__(self, extension_handle): 

56 """Returns the current value of the given extension handle.""" 

57 

58 _VerifyExtensionHandle(self._extended_message, extension_handle) 

59 

60 result = self._extended_message._fields.get(extension_handle) 

61 if result is not None: 

62 return result 

63 

64 if extension_handle.is_repeated: 

65 result = extension_handle._default_constructor(self._extended_message) 

66 elif extension_handle.cpp_type == FieldDescriptor.CPPTYPE_MESSAGE: 

67 message_type = extension_handle.message_type 

68 if not hasattr(message_type, '_concrete_class'): 

69 # pylint: disable=g-import-not-at-top 

70 from google.protobuf import message_factory 

71 message_factory.GetMessageClass(message_type) 

72 if not hasattr(extension_handle.message_type, '_concrete_class'): 

73 from google.protobuf import message_factory 

74 message_factory.GetMessageClass(extension_handle.message_type) 

75 result = extension_handle.message_type._concrete_class() 

76 try: 

77 result._SetListener(self._extended_message._listener_for_children) 

78 except ReferenceError: 

79 pass 

80 else: 

81 # Singular scalar -- just return the default without inserting into the 

82 # dict. 

83 return extension_handle.default_value 

84 

85 # Atomically check if another thread has preempted us and, if not, swap 

86 # in the new object we just created. If someone has preempted us, we 

87 # take that object and discard ours. 

88 # WARNING: We are relying on setdefault() being atomic. This is true 

89 # in CPython but we haven't investigated others. This warning appears 

90 # in several other locations in this file. 

91 result = self._extended_message._fields.setdefault( 

92 extension_handle, result) 

93 

94 return result 

95 

96 def __eq__(self, other): 

97 if not isinstance(other, self.__class__): 

98 return False 

99 

100 my_fields = self._extended_message.ListFields() 

101 other_fields = other._extended_message.ListFields() 

102 

103 # Get rid of non-extension fields. 

104 my_fields = [field for field in my_fields if field.is_extension] 

105 other_fields = [field for field in other_fields if field.is_extension] 

106 

107 return my_fields == other_fields 

108 

109 def __ne__(self, other): 

110 return not self == other 

111 

112 def __len__(self): 

113 fields = self._extended_message.ListFields() 

114 # Get rid of non-extension fields. 

115 extension_fields = [field for field in fields if field[0].is_extension] 

116 return len(extension_fields) 

117 

118 def __hash__(self): 

119 raise TypeError('unhashable object') 

120 

121 # Note that this is only meaningful for non-repeated, scalar extension 

122 # fields. Note also that we may have to call _Modified() when we do 

123 # successfully set a field this way, to set any necessary "has" bits in the 

124 # ancestors of the extended message. 

125 def __setitem__(self, extension_handle, value): 

126 """If extension_handle specifies a non-repeated, scalar extension 

127 field, sets the value of that field. 

128 """ 

129 

130 _VerifyExtensionHandle(self._extended_message, extension_handle) 

131 

132 if (extension_handle.is_repeated or 

133 extension_handle.cpp_type == FieldDescriptor.CPPTYPE_MESSAGE): 

134 raise TypeError( 

135 'Cannot assign to extension "%s" because it is a repeated or ' 

136 'composite type.' % extension_handle.full_name) 

137 

138 # It's slightly wasteful to lookup the type checker each time, 

139 # but we expect this to be a vanishingly uncommon case anyway. 

140 type_checker = type_checkers.GetTypeChecker(extension_handle) 

141 # pylint: disable=protected-access 

142 self._extended_message._fields[extension_handle] = ( 

143 type_checker.CheckValue(value)) 

144 self._extended_message._Modified() 

145 

146 def __delitem__(self, extension_handle): 

147 self._extended_message.ClearExtension(extension_handle) 

148 

149 def _FindExtensionByName(self, name): 

150 """Tries to find a known extension with the specified name. 

151 

152 Args: 

153 name: Extension full name. 

154 

155 Returns: 

156 Extension field descriptor. 

157 """ 

158 descriptor = self._extended_message.DESCRIPTOR 

159 extensions = descriptor.file.pool._extensions_by_name[descriptor] 

160 return extensions.get(name, None) 

161 

162 def _FindExtensionByNumber(self, number): 

163 """Tries to find a known extension with the field number. 

164 

165 Args: 

166 number: Extension field number. 

167 

168 Returns: 

169 Extension field descriptor. 

170 """ 

171 descriptor = self._extended_message.DESCRIPTOR 

172 extensions = descriptor.file.pool._extensions_by_number[descriptor] 

173 return extensions.get(number, None) 

174 

175 def __iter__(self): 

176 # Return a generator over the populated extension fields 

177 return (f[0] for f in self._extended_message.ListFields() 

178 if f[0].is_extension) 

179 

180 def __contains__(self, extension_handle): 

181 _VerifyExtensionHandle(self._extended_message, extension_handle) 

182 

183 if extension_handle not in self._extended_message._fields: 

184 return False 

185 

186 if extension_handle.is_repeated: 

187 return bool(self._extended_message._fields.get(extension_handle)) 

188 

189 if extension_handle.cpp_type == FieldDescriptor.CPPTYPE_MESSAGE: 

190 value = self._extended_message._fields.get(extension_handle) 

191 # pylint: disable=protected-access 

192 return value is not None and value._is_present_in_parent 

193 

194 return True