def list_cloudformation_stack_resources()

in Gems/AWSCore/Code/Tools/ResourceMappingTool/utils/aws_utils.py [0:0]

• 29 lines of code
• 10 McCabe index (conditional complexity)

def list_cloudformation_stack_resources(stack_name, region=None) -> List[BasicResourceAttributes]:
    cfn_client: BaseClient = _initialize_boto3_aws_client(AWSConstants.CLOUDFORMATION_SERVICE_NAME, region)
    try:
        cfn_paginator: Paginator = cfn_client.get_paginator(AWSConstants.CLOUDFORMATION_LIST_STACK_RESOURCES_API_NAME)
        resource_type_and_name: List[BasicResourceAttributes] = []

        starting_token: str = None
        while True:
            # list cloudformation stack resources with starting token and max items. StartingToken is used to mark
            # the starting point of the request; if None, it means request from start. MaxItems is used to define the
            # total number of resources requested in the page iterator.
            iterator: PageIterator = \
                cfn_paginator.paginate(StackName=stack_name,
                                       PaginationConfig={"MaxItems": _PAGINATION_MAX_ITEMS,
                                                         "StartingToken": starting_token})
            page: Dict[str, any]
            for page in iterator:
                # iterate through page iterator to fetch all resources
                resource: Dict[str, any]
                for resource in page["StackResourceSummaries"]:
                    if "ResourceType" in resource.keys() and "PhysicalResourceId" in resource.keys():
                        resource_type_and_name.append(BasicResourceAttributesBuilder()
                                                      .build_type(resource["ResourceType"])
                                                      .build_name_id(resource["PhysicalResourceId"])
                                                      .build())
            if iterator.resume_token is None:
                # when resume token is none, it means there is no more resources left
                break
            else:
                # setting starting token to resume token, then next request will query with proper starting point
                starting_token = iterator.resume_token
        return resource_type_and_name
    except ClientError as error:
        raise RuntimeError(error_messages.AWS_SERVICE_REQUEST_CLIENT_ERROR_MESSAGE.format(
            AWSConstants.CLOUDFORMATION_LIST_STACK_RESOURCES_API_NAME,
            error.response['Error']['Code'], error.response['Error']['Message']))