def update_extension_index()

in azdev/operations/extensions/__init__.py [0:0]

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

def update_extension_index(extensions):
    import re
    import tempfile

    from .util import get_ext_metadata, get_whl_from_url

    ext_repos = get_ext_repo_paths()
    index_path = next((x for x in find_files(ext_repos, 'index.json') if 'azure-cli-extensions' in x), None)
    if not index_path:
        raise CLIError("Unable to find 'index.json' in your extension repos. Have "
                       "you cloned 'azure-cli-extensions' and added it to you repo "
                       "sources with `azdev extension repo add`?")

    NAME_REGEX = r'.*/([^/]*)-\d+.\d+.\d+'

    for extension in extensions:
        # Get the URL
        extension = extension[extension.index('https'):]
        # Get extension WHL from URL
        if not extension.endswith('.whl') or not extension.startswith('https:'):
            raise CLIError('usage error: only URL to a WHL file currently supported.')

        # TODO: extend to consider other options
        ext_path = extension

        # Extract the extension name
        try:
            extension_name = re.findall(NAME_REGEX, ext_path)[0]
            extension_name = extension_name.replace('_', '-')
        except IndexError:
            raise CLIError('unable to parse extension name')

        # TODO: Update this!
        extensions_dir = tempfile.mkdtemp()
        ext_dir = tempfile.mkdtemp(dir=extensions_dir)
        whl_cache_dir = tempfile.mkdtemp()
        whl_cache = {}
        ext_file = get_whl_from_url(ext_path, extension_name, whl_cache_dir, whl_cache)

        with open(index_path, 'r') as infile:
            curr_index = json.loads(infile.read())

        entry = {
            'downloadUrl': ext_path,
            'sha256Digest': _get_sha256sum(ext_file),
            'filename': ext_path.split('/')[-1],
            'metadata': get_ext_metadata(ext_dir, ext_file, extension_name)
        }

        if extension_name not in curr_index['extensions'].keys():
            logger.info("Adding '%s' to index...", extension_name)
            curr_index['extensions'][extension_name] = [entry]
        else:
            logger.info("Updating '%s' in index...", extension_name)
            curr_index['extensions'][extension_name].append(entry)

        # update index and write back to file
        with open(os.path.join(index_path), 'w') as outfile:
            outfile.write(json.dumps(curr_index, indent=4, sort_keys=True))