def update_resource_spec()

in src/cfnlint/maintenance.py [0:0]

• 67 lines of code
• 12 McCabe index (conditional complexity)

def update_resource_spec(region, url, schema_cache):
    """ Update a single resource spec """
    filename = os.path.join(os.path.dirname(cfnlint.__file__), 'data/CloudSpecs/%s.json' % region)

    multiprocessing_logger = multiprocessing.log_to_stderr()

    multiprocessing_logger.debug('Downloading template %s into %s', url, filename)

    # Check to see if we already have the latest version, and if so stop
    if not url_has_newer_version(url):
        return

    spec_content = get_url_content(url, caching=True)

    multiprocessing_logger.debug(
        'A more recent version of %s was found, and will be downloaded to %s', url, filename)
    spec = json.loads(spec_content)

    # Patch the files
    spec = patch_spec(spec, 'all')
    spec = patch_spec(spec, region)

    # do each patch individually so we can ignore errors
    for patch in schema_cache:
        try:
            # since there could be patched in values to ValueTypes
            # Ref/GetAtt as an example.  So we want to add new
            # ValueTypes that don't exist
            for i_patch in patch:
                path_details = i_patch.get('path').split('/')
                if path_details[1] == 'ValueTypes':
                    if not spec.get('ValueTypes').get(path_details[2]):
                        spec['ValueTypes'][path_details[2]] = {}
            # do the patch
            jsonpatch.JsonPatch(patch).apply(spec, in_place=True)
        except jsonpatch.JsonPatchConflict:
            for i_patch in patch:
                path_details = i_patch.get('path').split('/')
                if path_details[1] == 'ValueTypes':
                    if not spec.get('ValueTypes').get(path_details[2]):
                        try:
                            del spec['ValueTypes'][path_details[2]]
                        except:  # pylint: disable=bare-except
                            pass
            LOGGER.debug('Patch (%s) not applied in region %s', patch, region)
        except jsonpatch.JsonPointerException:
            for i_patch in patch:
                path_details = i_patch.get('path').split('/')
                if path_details[1] == 'ValueTypes':
                    if not spec.get('ValueTypes').get(path_details[2]):
                        try:
                            del spec['ValueTypes'][path_details[2]]
                        except:  # pylint: disable=bare-except
                            pass
            # Debug as the parent element isn't supported in the region
            LOGGER.debug('Parent element not found for patch (%s) in region %s',
                         patch, region)

    botocore_cache = {}

    def search_and_replace_botocore_types(obj):
        if isinstance(obj, dict):
            new_obj = {}
            for key, value in obj.items():
                if key == 'botocore':
                    service_and_type = value.split('/')
                    service = '/'.join(service_and_type[:-1])
                    botocore_type = service_and_type[-1]
                    if service not in botocore_cache:
                        botocore_cache[service] = json.loads(get_url_content(
                            'https://raw.githubusercontent.com/boto/botocore/master/botocore/data/' + service + '/service-2.json'))
                    new_obj['AllowedValues'] = sorted(
                        botocore_cache[service]['shapes'][botocore_type]['enum'])
                else:
                    new_obj[key] = search_and_replace_botocore_types(value)
            return new_obj

        if isinstance(obj, list):
            new_list = []
            for item in obj:
                new_list.append(search_and_replace_botocore_types(item))
            return new_list

        return obj

    spec = search_and_replace_botocore_types(spec)

    with open(filename, 'w') as f:
        json.dump(spec, f, indent=2, sort_keys=True, separators=(',', ': '))