You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
168 lines
6.0 KiB
168 lines
6.0 KiB
# 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]
|