def _update_artifacttool()

in azure-devops/azext_devops/dev/common/artifacttool_updater.py [0:0]

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

def _update_artifacttool(uri, release_id):
    root = _compute_artifacttool_root()

    # Remove all existing releases. In the future we may maintain some old versions,
    # but right now we always delete them.
    if os.path.isdir(root):
        for item in os.listdir(root):
            path = os.path.join(root, item)
            if os.path.isdir(path):
                logger.debug("Trying to remove old release %s", item)
                shutil.rmtree(path, ignore_errors=True)   # Failing cleanup is not fatal

    with humanfriendly.Spinner(  # pylint: disable=no-member
            label="Downloading Universal Packages tooling ({})"
            .format(release_id), total=100, stream=sys.stderr) as spinner:
        spinner.step()
        logger.debug("Downloading ArtifactTool from %s", uri)

        # Make the request, determine the total size
        response = requests.get(uri, stream=True)
        content_length_header = response.headers['Content-Length'].strip()
        content_length = int(content_length_header)

        # Do the download, updating the progress bar
        content = io.BytesIO()
        bytes_so_far = 0
        for chunk in response.iter_content(chunk_size=1024 * 512):
            if chunk:
                content.write(chunk)
                bytes_so_far += len(chunk)
                spinner.step(100 * float(bytes_so_far) / float(content_length))

        # Extract the zip
        release_temp_dir = os.path.join(root, str(uuid.uuid4()))
        logger.debug("Extracting ArtifactTool to %s", release_temp_dir)
        f = zipfile.ZipFile(content)
        try:
            _mkdir_if_not_exist(release_temp_dir)
            f.extractall(path=release_temp_dir)

            # For Linux, ensure the executable bit is set on the binary "ArtifactTool" if it exists.
            # Python has a bug https://bugs.python.org/issue15795 where file permissions are not preserved.
            artifacttool_binary = os.path.join(release_temp_dir, "artifacttool")
            if os.path.exists(artifacttool_binary):
                artifacttool_stat = os.stat(artifacttool_binary)
                os.chmod(artifacttool_binary,
                         artifacttool_stat.st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)

            # Move the release into the real releases location
            release_dir = _compute_release_dir(release_id)
            if os.path.exists(release_dir):
                logger.warning(
                    "The Universal Packages tool already exists at the location %s. Skipping download.",
                    release_dir)
            else:
                logger.debug("Moving downloaded ArtifactTool from %s to %s", release_temp_dir, release_dir)
                # number of times to retry
                retries = 10
                for _ in range(retries - 1):
                    try:
                        os.rename(release_temp_dir, release_dir)
                        break
                    except BaseException as ex:  # pylint: disable=broad-except
                        logger.debug(
                            "An error occurred while renaming the Universal Packages tooling: %s. Retrying...", ex)
                        time.sleep(1)
                else:
                    os.rename(release_temp_dir, release_dir)
                logger.info("Downloaded Universal Packages tooling successfully")
        except BaseException as ex:  # pylint: disable=broad-except
            logger.error("An error occurred while extracting the Universal Packages tooling: %s", ex)
            logger.debug("Removing temporary directory %s", release_temp_dir)
            shutil.rmtree(release_temp_dir, ignore_errors=True)