# 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.importjmespathfrombotocoreimportxform_namefrom.paramsimportget_data_member
[docs]defall_not_none(iterable):""" Return True if all elements of the iterable are not None (or if the iterable is empty). This is like the built-in ``all``, except checks against None, so 0 and False are allowable values. """forelementiniterable:ifelementisNone:returnFalsereturnTrue
[docs]defbuild_identifiers(identifiers,parent,params=None,raw_response=None):""" Builds a mapping of identifier names to values based on the identifier source location, type, and target. Identifier values may be scalars or lists depending on the source type and location. :type identifiers: list :param identifiers: List of :py:class:`~boto3.resources.model.Parameter` definitions :type parent: ServiceResource :param parent: The resource instance to which this action is attached. :type params: dict :param params: Request parameters sent to the service. :type raw_response: dict :param raw_response: Low-level operation response. :rtype: list :return: An ordered list of ``(name, value)`` identifier tuples. """results=[]foridentifierinidentifiers:source=identifier.sourcetarget=identifier.targetifsource=='response':value=jmespath.search(identifier.path,raw_response)elifsource=='requestParameter':value=jmespath.search(identifier.path,params)elifsource=='identifier':value=getattr(parent,xform_name(identifier.name))elifsource=='data':# If this is a data member then it may incur a load# action before returning the value.value=get_data_member(parent,identifier.path)elifsource=='input':# This value is set by the user, so ignore it herecontinueelse:raiseNotImplementedError(f'Unsupported source type: {source}')results.append((xform_name(target),value))returnresults
[docs]defbuild_empty_response(search_path,operation_name,service_model):""" Creates an appropriate empty response for the type that is expected, based on the service model's shape type. For example, a value that is normally a list would then return an empty list. A structure would return an empty dict, and a number would return None. :type search_path: string :param search_path: JMESPath expression to search in the response :type operation_name: string :param operation_name: Name of the underlying service operation. :type service_model: :ref:`botocore.model.ServiceModel` :param service_model: The Botocore service model :rtype: dict, list, or None :return: An appropriate empty value """response=Noneoperation_model=service_model.operation_model(operation_name)shape=operation_model.output_shapeifsearch_path:# Walk the search path and find the final shape. For example, given# a path of ``foo.bar[0].baz``, we first find the shape for ``foo``,# then the shape for ``bar`` (ignoring the indexing), and finally# the shape for ``baz``.foriteminsearch_path.split('.'):item=item.strip('[0123456789]$')ifshape.type_name=='structure':shape=shape.members[item]elifshape.type_name=='list':shape=shape.memberelse:raiseNotImplementedError(f'Search path hits shape type {shape.type_name} from {item}')# Anything not handled here is set to Noneifshape.type_name=='structure':response={}elifshape.type_name=='list':response=[]elifshape.type_name=='map':response={}returnresponse
[docs]classRawHandler:""" A raw action response handler. This passed through the response dictionary, optionally after performing a JMESPath search if one has been defined for the action. :type search_path: string :param search_path: JMESPath expression to search in the response :rtype: dict :return: Service response """def__init__(self,search_path):self.search_path=search_pathdef__call__(self,parent,params,response):""" :type parent: ServiceResource :param parent: The resource instance to which this action is attached. :type params: dict :param params: Request parameters sent to the service. :type response: dict :param response: Low-level operation response. """# TODO: Remove the '$' check after JMESPath supports itifself.search_pathandself.search_path!='$':response=jmespath.search(self.search_path,response)returnresponse
[docs]classResourceHandler:""" Creates a new resource or list of new resources from the low-level response based on the given response resource definition. :type search_path: string :param search_path: JMESPath expression to search in the response :type factory: ResourceFactory :param factory: The factory that created the resource class to which this action is attached. :type resource_model: :py:class:`~boto3.resources.model.ResponseResource` :param resource_model: Response resource model. :type service_context: :py:class:`~boto3.utils.ServiceContext` :param service_context: Context about the AWS service :type operation_name: string :param operation_name: Name of the underlying service operation, if it exists. :rtype: ServiceResource or list :return: New resource instance(s). """def__init__(self,search_path,factory,resource_model,service_context,operation_name=None,):self.search_path=search_pathself.factory=factoryself.resource_model=resource_modelself.operation_name=operation_nameself.service_context=service_contextdef__call__(self,parent,params,response):""" :type parent: ServiceResource :param parent: The resource instance to which this action is attached. :type params: dict :param params: Request parameters sent to the service. :type response: dict :param response: Low-level operation response. """resource_name=self.resource_model.typejson_definition=self.service_context.resource_json_definitions.get(resource_name)# Load the new resource class that will result from this action.resource_cls=self.factory.load_from_definition(resource_name=resource_name,single_resource_json_definition=json_definition,service_context=self.service_context,)raw_response=responsesearch_response=None# Anytime a path is defined, it means the response contains the# resource's attributes, so resource_data gets set here. It# eventually ends up in resource.meta.data, which is where# the attribute properties look for data.ifself.search_path:search_response=jmespath.search(self.search_path,raw_response)# First, we parse all the identifiers, then create the individual# response resources using them. Any identifiers that are lists# will have one item consumed from the front of the list for each# resource that is instantiated. Items which are not a list will# be set as the same value on each new resource instance.identifiers=dict(build_identifiers(self.resource_model.identifiers,parent,params,raw_response))# If any of the identifiers is a list, then the response is pluralplural=[vforvinidentifiers.values()ifisinstance(v,list)]ifplural:response=[]# The number of items in an identifier that is a list will# determine how many resource instances to create.foriinrange(len(plural[0])):# Response item data is *only* available if a search path# was given. This prevents accidentally loading unrelated# data that may be in the response.response_item=Noneifsearch_response:response_item=search_response[i]response.append(self.handle_response_item(resource_cls,parent,identifiers,response_item))elifall_not_none(identifiers.values()):# All identifiers must always exist, otherwise the resource# cannot be instantiated.response=self.handle_response_item(resource_cls,parent,identifiers,search_response)else:# The response should be empty, but that may mean an# empty dict, list, or None based on whether we make# a remote service call and what shape it is expected# to return.response=Noneifself.operation_nameisnotNone:# A remote service call was made, so try and determine# its shape.response=build_empty_response(self.search_path,self.operation_name,self.service_context.service_model,)returnresponse
[docs]defhandle_response_item(self,resource_cls,parent,identifiers,resource_data):""" Handles the creation of a single response item by setting parameters and creating the appropriate resource instance. :type resource_cls: ServiceResource subclass :param resource_cls: The resource class to instantiate. :type parent: ServiceResource :param parent: The resource instance to which this action is attached. :type identifiers: dict :param identifiers: Map of identifier names to value or values. :type resource_data: dict or None :param resource_data: Data for resource attributes. :rtype: ServiceResource :return: New resource instance. """kwargs={'client':parent.meta.client,}forname,valueinidentifiers.items():# If value is a list, then consume the next itemifisinstance(value,list):value=value.pop(0)kwargs[name]=valueresource=resource_cls(**kwargs)ifresource_dataisnotNone:resource.meta.data=resource_datareturnresource