in source/code/services/aws_service.py [0:0]
def describe(self, service_resource, region=None, tags=False, tags_as_dict=None, as_tuple=None,
select=None, filter_func=None, context=None, select_on_tag=None, tag_roles=None, **describe_args):
"""
This method is used to retrieve service resources, specified by their name, from a service
:param filter_func: function for additional filtering of resources
:param service_resource: Name of the service resource, not case sensitive, use camel or snake case
:param region: Region from where resources are retrieved, if None then the current region is used
:param tags: Set to True to return tags with the resource
:param tags_as_dict: Set to True to return tags as python dictionaries
:param as_tuple: Set to true to return results as immutable named dictionaries instead of dictionaries
:param select: JMES path to select resources and select/transform attributes of returned resources
:param select_on_tag: only include resources that have a tag with this name
:param tag_roles: optional roles used to assume to select tags for a resource as this may be required by shared resources
from another account
:param describe_args: Parameters passed to the boto "describe" function
:param context: Lambda context
:return: Service resources of the specified resource type for the service.
"""
def use_tuple():
"""
Tests if resources should be returned as named tuples
:return: True for tuples, False for dictionaries
"""
return (as_tuple is not None and as_tuple) or (as_tuple is None and self._as_tuple)
def tags_as_dictionary():
"""
Tests if tags should be returned as python dictionaries
:return: True for dictionaries, False for original tag format
"""
return tags_as_dict if tags_as_dict is not None else self._tags_as_dict
# normalize resource name
self._resource_name = self._get_resource_name(service_resource)
# get the name of the boto3 method to retrieve this resource type
describe_func_name = self.describe_resources_function_name(self._resource_name)
# get additional parameters for boto describe method and map parameter names
if describe_args is None:
function_args = {}
else:
function_args = self._map_describe_function_parameters(self._resource_name, describe_args)
# get method from boto service client
if self._service_retry_strategy is not None:
method_names = [describe_func_name]
describe_func_name = describe_func_name + boto_retry.DEFAULT_SUFFIX
else:
method_names = None
client = self.service_client(region=region, method_names=method_names)
describe_func = getattr(client, describe_func_name, None)
if describe_func is None:
raise_value_error(ERR_NO_BOTO_SERVICE_METHOD, self.service_name, describe_func_name)
self._cached_tags = None
next_token = self._next_token_result_name(self._resource_name)
self._tags_as_dict = tags_as_dictionary()
self._use_tuple = use_tuple()
self._describe_args = describe_args
self._select_on_tag = select_on_tag
self._tags = tags
self._tag_roles = tag_roles
self._context = context
done = False
while not done:
# call boto method to retrieve until no more resources are retrieved
try:
resp = describe_func(**function_args)
except Exception as ex:
expected_exceptions = describe_args.get(boto_retry.EXPECTED_EXCEPTIONS, [])
if type(ex).__name__ in expected_exceptions or getattr(ex, "response", {}).get("Error", {}) \
.get("Code", "") in expected_exceptions:
done = True
continue
else:
raise ex
# extract resources from result and transform to requested output format
resources_data = self._extract_resources(resp=resp, select=select)
self._use_cached_tags = self.__class__.use_cached_tags(self._resource_name, len(resources_data))
for obj in resources_data:
if filter_func is not None and not filter_func(obj):
continue
# annotate additional account and region attributes
obj["AwsAccount"] = self.aws_account
obj["Region"] = self.service_client(region).meta.region_name if self.is_regional() else None
obj["Service"] = self.service_name
obj["ResourceTypeName"] = self._resource_name
# yield the transformed resource
transformed = self._transform_returned_resource(self.service_client(region=region), resource=obj)
if select_on_tag is None or select_on_tag in transformed.get("Tags", {}):
yield transformed
# if there are set the continuation token parameter for the next call to the value of the results continuation token
# test if more resources are available
if next_token in resp and resp[next_token] not in ["", False, None]:
self.set_continuation_call_parameters(function_args, next_token, resp)
else:
# all resources retrieved
done = True