Spaces:
Paused
Paused
| # Copyright 2014 Amazon.com, Inc. or its affiliates. All Rights Reserved. | |
| # | |
| # Licensed under the Apache License, Version 2.0 (the "License"). You | |
| # may not use this file except in compliance with the License. A copy of | |
| # the License is located at | |
| # | |
| # https://aws.amazon.com/apache2.0/ | |
| # | |
| # or in the "license" file accompanying this file. This file is | |
| # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF | |
| # ANY KIND, either express or implied. See the License for the specific | |
| # language governing permissions and limitations under the License. | |
| import re | |
| import jmespath | |
| from botocore import xform_name | |
| from ..exceptions import ResourceLoadException | |
| INDEX_RE = re.compile(r'\[(.*)\]$') | |
| def get_data_member(parent, path): | |
| """ | |
| Get a data member from a parent using a JMESPath search query, | |
| loading the parent if required. If the parent cannot be loaded | |
| and no data is present then an exception is raised. | |
| :type parent: ServiceResource | |
| :param parent: The resource instance to which contains data we | |
| are interested in. | |
| :type path: string | |
| :param path: The JMESPath expression to query | |
| :raises ResourceLoadException: When no data is present and the | |
| resource cannot be loaded. | |
| :returns: The queried data or ``None``. | |
| """ | |
| # Ensure the parent has its data loaded, if possible. | |
| if parent.meta.data is None: | |
| if hasattr(parent, 'load'): | |
| parent.load() | |
| else: | |
| raise ResourceLoadException( | |
| f'{parent.__class__.__name__} has no load method!' | |
| ) | |
| return jmespath.search(path, parent.meta.data) | |
| def create_request_parameters(parent, request_model, params=None, index=None): | |
| """ | |
| Handle request parameters that can be filled in from identifiers, | |
| resource data members or constants. | |
| By passing ``params``, you can invoke this method multiple times and | |
| build up a parameter dict over time, which is particularly useful | |
| for reverse JMESPath expressions that append to lists. | |
| :type parent: ServiceResource | |
| :param parent: The resource instance to which this action is attached. | |
| :type request_model: :py:class:`~boto3.resources.model.Request` | |
| :param request_model: The action request model. | |
| :type params: dict | |
| :param params: If set, then add to this existing dict. It is both | |
| edited in-place and returned. | |
| :type index: int | |
| :param index: The position of an item within a list | |
| :rtype: dict | |
| :return: Pre-filled parameters to be sent to the request operation. | |
| """ | |
| if params is None: | |
| params = {} | |
| for param in request_model.params: | |
| source = param.source | |
| target = param.target | |
| if source == 'identifier': | |
| # Resource identifier, e.g. queue.url | |
| value = getattr(parent, xform_name(param.name)) | |
| elif source == 'data': | |
| # If this is a data member then it may incur a load | |
| # action before returning the value. | |
| value = get_data_member(parent, param.path) | |
| elif source in ['string', 'integer', 'boolean']: | |
| # These are hard-coded values in the definition | |
| value = param.value | |
| elif source == 'input': | |
| # This is provided by the user, so ignore it here | |
| continue | |
| else: | |
| raise NotImplementedError(f'Unsupported source type: {source}') | |
| build_param_structure(params, target, value, index) | |
| return params | |
| def build_param_structure(params, target, value, index=None): | |
| """ | |
| This method provides a basic reverse JMESPath implementation that | |
| lets you go from a JMESPath-like string to a possibly deeply nested | |
| object. The ``params`` are mutated in-place, so subsequent calls | |
| can modify the same element by its index. | |
| >>> build_param_structure(params, 'test[0]', 1) | |
| >>> print(params) | |
| {'test': [1]} | |
| >>> build_param_structure(params, 'foo.bar[0].baz', 'hello world') | |
| >>> print(params) | |
| {'test': [1], 'foo': {'bar': [{'baz': 'hello, world'}]}} | |
| """ | |
| pos = params | |
| parts = target.split('.') | |
| # First, split into parts like 'foo', 'bar[0]', 'baz' and process | |
| # each piece. It can either be a list or a dict, depending on if | |
| # an index like `[0]` is present. We detect this via a regular | |
| # expression, and keep track of where we are in params via the | |
| # pos variable, walking down to the last item. Once there, we | |
| # set the value. | |
| for i, part in enumerate(parts): | |
| # Is it indexing an array? | |
| result = INDEX_RE.search(part) | |
| if result: | |
| if result.group(1): | |
| if result.group(1) == '*': | |
| part = part[:-3] | |
| else: | |
| # We have an explicit index | |
| index = int(result.group(1)) | |
| part = part[: -len(str(index) + '[]')] | |
| else: | |
| # Index will be set after we know the proper part | |
| # name and that it's a list instance. | |
| index = None | |
| part = part[:-2] | |
| if part not in pos or not isinstance(pos[part], list): | |
| pos[part] = [] | |
| # This means we should append, e.g. 'foo[]' | |
| if index is None: | |
| index = len(pos[part]) | |
| while len(pos[part]) <= index: | |
| # Assume it's a dict until we set the final value below | |
| pos[part].append({}) | |
| # Last item? Set the value, otherwise set the new position | |
| if i == len(parts) - 1: | |
| pos[part][index] = value | |
| else: | |
| # The new pos is the *item* in the array, not the array! | |
| pos = pos[part][index] | |
| else: | |
| if part not in pos: | |
| pos[part] = {} | |
| # Last item? Set the value, otherwise set the new position | |
| if i == len(parts) - 1: | |
| pos[part] = value | |
| else: | |
| pos = pos[part] | |