We use cookies and similar tools to enhance your experience, provide our services, deliver relevant advertising, and make improvements. Approved third parties also use these tools to help us deliver advertising and provide certain site features.
Customize cookie preferences
We use cookies and similar tools (collectively, "cookies") for the following purposes.
Essential
Essential cookies are necessary to provide our site and services and cannot be deactivated. They are usually set in response to your actions on the site, such as setting your privacy preferences, signing in, or filling in forms.
Performance
Performance cookies provide anonymous statistics about how customers navigate our site so we can improve site experience and performance. Approved third parties may perform analytics on our behalf, but they cannot use the data for their own purposes.
Allowed
Functional
Functional cookies help us provide useful site features, remember your preferences, and display relevant content. Approved third parties may set these cookies to provide certain site features. If you do not allow these cookies, then some or all of these services may not function properly.
Allowed
Advertising
Advertising cookies may be set through our site by us or our advertising partners and help us deliver relevant marketing content. If you do not allow these cookies, you will experience less relevant advertising.
Allowed
Blocking some types of cookies may impact your experience of our sites. You may review and change your choices at any time by clicking Cookie preferences in the footer of this site. We and selected third-parties use cookies or similar technologies as specified in the AWS Cookie Notice.
# 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