Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/boto3/resources/response.py: 14%
92 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-08 06:51 +0000
1# Copyright 2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
2#
3# Licensed under the Apache License, Version 2.0 (the "License"). You
4# may not use this file except in compliance with the License. A copy of
5# the License is located at
6#
7# https://aws.amazon.com/apache2.0/
8#
9# or in the "license" file accompanying this file. This file is
10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
11# ANY KIND, either express or implied. See the License for the specific
12# language governing permissions and limitations under the License.
14import jmespath
15from botocore import xform_name
17from .params import get_data_member
20def all_not_none(iterable):
21 """
22 Return True if all elements of the iterable are not None (or if the
23 iterable is empty). This is like the built-in ``all``, except checks
24 against None, so 0 and False are allowable values.
25 """
26 for element in iterable:
27 if element is None:
28 return False
29 return True
32def build_identifiers(identifiers, parent, params=None, raw_response=None):
33 """
34 Builds a mapping of identifier names to values based on the
35 identifier source location, type, and target. Identifier
36 values may be scalars or lists depending on the source type
37 and location.
39 :type identifiers: list
40 :param identifiers: List of :py:class:`~boto3.resources.model.Parameter`
41 definitions
42 :type parent: ServiceResource
43 :param parent: The resource instance to which this action is attached.
44 :type params: dict
45 :param params: Request parameters sent to the service.
46 :type raw_response: dict
47 :param raw_response: Low-level operation response.
48 :rtype: list
49 :return: An ordered list of ``(name, value)`` identifier tuples.
50 """
51 results = []
53 for identifier in identifiers:
54 source = identifier.source
55 target = identifier.target
57 if source == 'response':
58 value = jmespath.search(identifier.path, raw_response)
59 elif source == 'requestParameter':
60 value = jmespath.search(identifier.path, params)
61 elif source == 'identifier':
62 value = getattr(parent, xform_name(identifier.name))
63 elif source == 'data':
64 # If this is a data member then it may incur a load
65 # action before returning the value.
66 value = get_data_member(parent, identifier.path)
67 elif source == 'input':
68 # This value is set by the user, so ignore it here
69 continue
70 else:
71 raise NotImplementedError(f'Unsupported source type: {source}')
73 results.append((xform_name(target), value))
75 return results
78def build_empty_response(search_path, operation_name, service_model):
79 """
80 Creates an appropriate empty response for the type that is expected,
81 based on the service model's shape type. For example, a value that
82 is normally a list would then return an empty list. A structure would
83 return an empty dict, and a number would return None.
85 :type search_path: string
86 :param search_path: JMESPath expression to search in the response
87 :type operation_name: string
88 :param operation_name: Name of the underlying service operation.
89 :type service_model: :ref:`botocore.model.ServiceModel`
90 :param service_model: The Botocore service model
91 :rtype: dict, list, or None
92 :return: An appropriate empty value
93 """
94 response = None
96 operation_model = service_model.operation_model(operation_name)
97 shape = operation_model.output_shape
99 if search_path:
100 # Walk the search path and find the final shape. For example, given
101 # a path of ``foo.bar[0].baz``, we first find the shape for ``foo``,
102 # then the shape for ``bar`` (ignoring the indexing), and finally
103 # the shape for ``baz``.
104 for item in search_path.split('.'):
105 item = item.strip('[0123456789]$')
107 if shape.type_name == 'structure':
108 shape = shape.members[item]
109 elif shape.type_name == 'list':
110 shape = shape.member
111 else:
112 raise NotImplementedError(
113 'Search path hits shape type {} from {}'.format(
114 shape.type_name, item
115 )
116 )
118 # Anything not handled here is set to None
119 if shape.type_name == 'structure':
120 response = {}
121 elif shape.type_name == 'list':
122 response = []
123 elif shape.type_name == 'map':
124 response = {}
126 return response
129class RawHandler:
130 """
131 A raw action response handler. This passed through the response
132 dictionary, optionally after performing a JMESPath search if one
133 has been defined for the action.
135 :type search_path: string
136 :param search_path: JMESPath expression to search in the response
137 :rtype: dict
138 :return: Service response
139 """
141 def __init__(self, search_path):
142 self.search_path = search_path
144 def __call__(self, parent, params, response):
145 """
146 :type parent: ServiceResource
147 :param parent: The resource instance to which this action is attached.
148 :type params: dict
149 :param params: Request parameters sent to the service.
150 :type response: dict
151 :param response: Low-level operation response.
152 """
153 # TODO: Remove the '$' check after JMESPath supports it
154 if self.search_path and self.search_path != '$':
155 response = jmespath.search(self.search_path, response)
157 return response
160class ResourceHandler:
161 """
162 Creates a new resource or list of new resources from the low-level
163 response based on the given response resource definition.
165 :type search_path: string
166 :param search_path: JMESPath expression to search in the response
168 :type factory: ResourceFactory
169 :param factory: The factory that created the resource class to which
170 this action is attached.
172 :type resource_model: :py:class:`~boto3.resources.model.ResponseResource`
173 :param resource_model: Response resource model.
175 :type service_context: :py:class:`~boto3.utils.ServiceContext`
176 :param service_context: Context about the AWS service
178 :type operation_name: string
179 :param operation_name: Name of the underlying service operation, if it
180 exists.
182 :rtype: ServiceResource or list
183 :return: New resource instance(s).
184 """
186 def __init__(
187 self,
188 search_path,
189 factory,
190 resource_model,
191 service_context,
192 operation_name=None,
193 ):
194 self.search_path = search_path
195 self.factory = factory
196 self.resource_model = resource_model
197 self.operation_name = operation_name
198 self.service_context = service_context
200 def __call__(self, parent, params, response):
201 """
202 :type parent: ServiceResource
203 :param parent: The resource instance to which this action is attached.
204 :type params: dict
205 :param params: Request parameters sent to the service.
206 :type response: dict
207 :param response: Low-level operation response.
208 """
209 resource_name = self.resource_model.type
210 json_definition = self.service_context.resource_json_definitions.get(
211 resource_name
212 )
214 # Load the new resource class that will result from this action.
215 resource_cls = self.factory.load_from_definition(
216 resource_name=resource_name,
217 single_resource_json_definition=json_definition,
218 service_context=self.service_context,
219 )
220 raw_response = response
221 search_response = None
223 # Anytime a path is defined, it means the response contains the
224 # resource's attributes, so resource_data gets set here. It
225 # eventually ends up in resource.meta.data, which is where
226 # the attribute properties look for data.
227 if self.search_path:
228 search_response = jmespath.search(self.search_path, raw_response)
230 # First, we parse all the identifiers, then create the individual
231 # response resources using them. Any identifiers that are lists
232 # will have one item consumed from the front of the list for each
233 # resource that is instantiated. Items which are not a list will
234 # be set as the same value on each new resource instance.
235 identifiers = dict(
236 build_identifiers(
237 self.resource_model.identifiers, parent, params, raw_response
238 )
239 )
241 # If any of the identifiers is a list, then the response is plural
242 plural = [v for v in identifiers.values() if isinstance(v, list)]
244 if plural:
245 response = []
247 # The number of items in an identifier that is a list will
248 # determine how many resource instances to create.
249 for i in range(len(plural[0])):
250 # Response item data is *only* available if a search path
251 # was given. This prevents accidentally loading unrelated
252 # data that may be in the response.
253 response_item = None
254 if search_response:
255 response_item = search_response[i]
256 response.append(
257 self.handle_response_item(
258 resource_cls, parent, identifiers, response_item
259 )
260 )
261 elif all_not_none(identifiers.values()):
262 # All identifiers must always exist, otherwise the resource
263 # cannot be instantiated.
264 response = self.handle_response_item(
265 resource_cls, parent, identifiers, search_response
266 )
267 else:
268 # The response should be empty, but that may mean an
269 # empty dict, list, or None based on whether we make
270 # a remote service call and what shape it is expected
271 # to return.
272 response = None
273 if self.operation_name is not None:
274 # A remote service call was made, so try and determine
275 # its shape.
276 response = build_empty_response(
277 self.search_path,
278 self.operation_name,
279 self.service_context.service_model,
280 )
282 return response
284 def handle_response_item(
285 self, resource_cls, parent, identifiers, resource_data
286 ):
287 """
288 Handles the creation of a single response item by setting
289 parameters and creating the appropriate resource instance.
291 :type resource_cls: ServiceResource subclass
292 :param resource_cls: The resource class to instantiate.
293 :type parent: ServiceResource
294 :param parent: The resource instance to which this action is attached.
295 :type identifiers: dict
296 :param identifiers: Map of identifier names to value or values.
297 :type resource_data: dict or None
298 :param resource_data: Data for resource attributes.
299 :rtype: ServiceResource
300 :return: New resource instance.
301 """
302 kwargs = {
303 'client': parent.meta.client,
304 }
306 for name, value in identifiers.items():
307 # If value is a list, then consume the next item
308 if isinstance(value, list):
309 value = value.pop(0)
311 kwargs[name] = value
313 resource = resource_cls(**kwargs)
315 if resource_data is not None:
316 resource.meta.data = resource_data
318 return resource