Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/google/protobuf/internal/extension_dict.py: 20%
83 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-08 06:40 +0000
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-08 06:40 +0000
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
8"""Contains _ExtensionDict class to represent extensions.
9"""
11from google.protobuf.internal import type_checkers
12from google.protobuf.descriptor import FieldDescriptor
15def _VerifyExtensionHandle(message, extension_handle):
16 """Verify that the given extension handle is valid."""
18 if not isinstance(extension_handle, FieldDescriptor):
19 raise KeyError('HasExtension() expects an extension handle, got: %s' %
20 extension_handle)
22 if not extension_handle.is_extension:
23 raise KeyError('"%s" is not an extension.' % extension_handle.full_name)
25 if not extension_handle.containing_type:
26 raise KeyError('"%s" is missing a containing_type.'
27 % extension_handle.full_name)
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))
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):
42 """Dict-like container for Extension fields on proto instances.
44 Note that in all cases we expect extension handles to be
45 FieldDescriptors.
46 """
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
55 def __getitem__(self, extension_handle):
56 """Returns the current value of the given extension handle."""
58 _VerifyExtensionHandle(self._extended_message, extension_handle)
60 result = self._extended_message._fields.get(extension_handle)
61 if result is not None:
62 return result
64 if extension_handle.label == FieldDescriptor.LABEL_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
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)
94 return result
96 def __eq__(self, other):
97 if not isinstance(other, self.__class__):
98 return False
100 my_fields = self._extended_message.ListFields()
101 other_fields = other._extended_message.ListFields()
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]
107 return my_fields == other_fields
109 def __ne__(self, other):
110 return not self == other
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)
118 def __hash__(self):
119 raise TypeError('unhashable object')
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 """
130 _VerifyExtensionHandle(self._extended_message, extension_handle)
132 if (extension_handle.label == FieldDescriptor.LABEL_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)
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()
146 def __delitem__(self, extension_handle):
147 self._extended_message.ClearExtension(extension_handle)
149 def _FindExtensionByName(self, name):
150 """Tries to find a known extension with the specified name.
152 Args:
153 name: Extension full name.
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)
162 def _FindExtensionByNumber(self, number):
163 """Tries to find a known extension with the field number.
165 Args:
166 number: Extension field number.
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)
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)
180 def __contains__(self, extension_handle):
181 _VerifyExtensionHandle(self._extended_message, extension_handle)
183 if extension_handle not in self._extended_message._fields:
184 return False
186 if extension_handle.label == FieldDescriptor.LABEL_REPEATED:
187 return bool(self._extended_message._fields.get(extension_handle))
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
194 return True