Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use extract_resource_attribute method in localstack

Best Python code snippet using localstack_python

template_deployer.py

Source:template_deployer.py

...278        if not list(filter(lambda marker, e=e: marker in str(e), markers)):279            LOGGER.warning('Unexpected error retrieving details for resource %s: %s %s - %s %s' %280                (resource_type, e, traceback.format_exc(), resource, resource_status))281    return None282def extract_resource_attribute(resource_type, resource, attribute):283    LOGGER.debug('Extract resource attribute: %s %s' % (resource_type, attribute))284    # extract resource specific attributes285    if resource_type == 'Lambda::Function':286        actual_attribute = 'FunctionArn' if attribute == 'Arn' else attribute287        return resource['Configuration'][actual_attribute]288    elif resource_type == 'DynamoDB::Table':289        actual_attribute = 'LatestStreamArn' if attribute == 'StreamArn' else attribute290        value = resource['Table'].get(actual_attribute)291        return value292    elif resource_type == 'ApiGateway::RestApi':293        if attribute == 'PhysicalResourceId':294            return resource['id']295        if attribute == 'RootResourceId':296            resources = aws_stack.connect_to_service('apigateway').get_resources(restApiId=resource['id'])['items']297            for res in resources:298                if res['path'] == '/' and not res.get('parentId'):299                    return res['id']300    elif resource_type == 'ApiGateway::Resource':301        if attribute == 'PhysicalResourceId':302            return resource['id']303    return resource.get(attribute)304def resolve_ref(stack_name, ref, resources, attribute):305    LOGGER.debug('Resolving ref %s - %s' % (ref, attribute))306    if ref == 'AWS::Region':307        return DEFAULT_REGION308    resource_status = describe_stack_resources(stack_name, ref)[0]309    attr_value = resource_status.get(attribute)310    if attr_value not in [None, '']:311        return attr_value312    # fetch resource details313    resource = resources.get(ref)314    resource_new = retrieve_resource_details(ref, resource_status, resources, stack_name)315    if not resource_new:316        return317    resource_type = get_resource_type(resource)318    result = extract_resource_attribute(resource_type, resource_new, attribute)319    if not result:320        LOGGER.warning('Unable to extract reference attribute %s from resource: %s' % (attribute, resource_new))321    return result322def resolve_refs_recursively(stack_name, value, resources):323    if isinstance(value, dict):324        if len(value) == 1 and 'Ref' in value:325            return resolve_ref(stack_name, value['Ref'],326                resources, attribute='PhysicalResourceId')327        elif len(value) == 1 and 'Fn::GetAtt' in value:328            return resolve_ref(stack_name, value['Fn::GetAtt'][0],329                resources, attribute=value['Fn::GetAtt'][1])330        else:331            for key, val in iteritems(value):332                value[key] = resolve_refs_recursively(stack_name, val, resources)...

34107_template_deployer.py

Source:34107_template_deployer.py

...278        if not list(filter(lambda marker, e=e: marker in str(e), markers)):279            LOGGER.warning('Unexpected error retrieving details for resource %s: %s %s - %s %s' %280                (resource_type, e, traceback.format_exc(), resource, resource_status))281    return None282def extract_resource_attribute(resource_type, resource, attribute):283    LOGGER.debug('Extract resource attribute: %s %s' % (resource_type, attribute))284    # extract resource specific attributes285    if resource_type == 'Lambda::Function':286        actual_attribute = 'FunctionArn' if attribute == 'Arn' else attribute287        return resource['Configuration'][actual_attribute]288    elif resource_type == 'DynamoDB::Table':289        actual_attribute = 'LatestStreamArn' if attribute == 'StreamArn' else attribute290        value = resource['Table'].get(actual_attribute)291        return value292    elif resource_type == 'ApiGateway::RestApi':293        if attribute == 'PhysicalResourceId':294            return resource['id']295        if attribute == 'RootResourceId':296            resources = aws_stack.connect_to_service('apigateway').get_resources(restApiId=resource['id'])['items']297            for res in resources:298                if res['path'] == '/' and not res.get('parentId'):299                    return res['id']300    elif resource_type == 'ApiGateway::Resource':301        if attribute == 'PhysicalResourceId':302            return resource['id']303    return resource.get(attribute)304def resolve_ref(stack_name, ref, resources, attribute):305    LOGGER.debug('Resolving ref %s - %s' % (ref, attribute))306    if ref == 'AWS::Region':307        return DEFAULT_REGION308    resource_status = describe_stack_resources(stack_name, ref)[0]309    attr_value = resource_status.get(attribute)310    if attr_value not in [None, '']:311        return attr_value312    # fetch resource details313    resource = resources.get(ref)314    resource_new = retrieve_resource_details(ref, resource_status, resources, stack_name)315    if not resource_new:316        return317    resource_type = get_resource_type(resource)318    result = extract_resource_attribute(resource_type, resource_new, attribute)319    if not result:320        LOGGER.warning('Unable to extract reference attribute %s from resource: %s' % (attribute, resource_new))321    return result322def resolve_refs_recursively(stack_name, value, resources):323    if isinstance(value, dict):324        if len(value) == 1 and 'Ref' in value:325            return resolve_ref(stack_name, value['Ref'],326                resources, attribute='PhysicalResourceId')327        elif len(value) == 1 and 'Fn::GetAtt' in value:328            return resolve_ref(stack_name, value['Fn::GetAtt'][0],329                resources, attribute=value['Fn::GetAtt'][1])330        else:331            for key, val in iteritems(value):332                value[key] = resolve_refs_recursively(stack_name, val, resources)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.