annotate env/lib/python3.9/site-packages/virtualenv/seed/wheels/acquire.py @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"

"""Bootstrap"""
from __future__ import absolute_import, unicode_literals

import logging
import sys
from operator import eq, lt

from virtualenv.util.path import Path
from virtualenv.util.six import ensure_str
from virtualenv.util.subprocess import Popen, subprocess

from .bundle import from_bundle
from .util import Version, Wheel, discover_wheels


def get_wheel(distribution, version, for_py_version, search_dirs, download, app_data, do_periodic_update, env):
    """
    Get a wheel with the given distribution-version-for_py_version trio, by using the extra search dir + download
    """
    # not all wheels are compatible with all python versions, so we need to py version qualify it
    # 1. acquire from bundle
    wheel = from_bundle(distribution, version, for_py_version, search_dirs, app_data, do_periodic_update, env)

    # 2. download from the internet
    if version not in Version.non_version and download:
        wheel = download_wheel(
            distribution=distribution,
            version_spec=Version.as_version_spec(version),
            for_py_version=for_py_version,
            search_dirs=search_dirs,
            app_data=app_data,
            to_folder=app_data.house,
            env=env,
        )
    return wheel


def download_wheel(distribution, version_spec, for_py_version, search_dirs, app_data, to_folder, env):
    to_download = "{}{}".format(distribution, version_spec or "")
    logging.debug("download wheel %s %s to %s", to_download, for_py_version, to_folder)
    cmd = [
        sys.executable,
        "-m",
        "pip",
        "download",
        "--progress-bar",
        "off",
        "--disable-pip-version-check",
        "--only-binary=:all:",
        "--no-deps",
        "--python-version",
        for_py_version,
        "-d",
        str(to_folder),
        to_download,
    ]
    # pip has no interface in python - must be a new sub-process
    env = pip_wheel_env_run(search_dirs, app_data, env)
    process = Popen(cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
    out, err = process.communicate()
    if process.returncode != 0:
        kwargs = {"output": out}
        if sys.version_info < (3, 5):
            kwargs["output"] += err
        else:
            kwargs["stderr"] = err
        raise subprocess.CalledProcessError(process.returncode, cmd, **kwargs)
    result = _find_downloaded_wheel(distribution, version_spec, for_py_version, to_folder, out)
    logging.debug("downloaded wheel %s", result.name)
    return result


def _find_downloaded_wheel(distribution, version_spec, for_py_version, to_folder, out):
    for line in out.splitlines():
        line = line.lstrip()
        for marker in ("Saved ", "File was already downloaded "):
            if line.startswith(marker):
                return Wheel(Path(line[len(marker) :]).absolute())
    # if for some reason the output does not match fallback to latest version with that spec
    return find_compatible_in_house(distribution, version_spec, for_py_version, to_folder)


def find_compatible_in_house(distribution, version_spec, for_py_version, in_folder):
    wheels = discover_wheels(in_folder, distribution, None, for_py_version)
    start, end = 0, len(wheels)
    if version_spec is not None:
        if version_spec.startswith("<"):
            from_pos, op = 1, lt
        elif version_spec.startswith("=="):
            from_pos, op = 2, eq
        else:
            raise ValueError(version_spec)
        version = Wheel.as_version_tuple(version_spec[from_pos:])
        start = next((at for at, w in enumerate(wheels) if op(w.version_tuple, version)), len(wheels))

    return None if start == end else wheels[start]


def pip_wheel_env_run(search_dirs, app_data, env):
    for_py_version = "{}.{}".format(*sys.version_info[0:2])
    env = env.copy()
    env.update(
        {
            ensure_str(k): str(v)  # python 2 requires these to be string only (non-unicode)
            for k, v in {"PIP_USE_WHEEL": "1", "PIP_USER": "0", "PIP_NO_INPUT": "1"}.items()
        },
    )
    wheel = get_wheel(
        distribution="pip",
        version=None,
        for_py_version=for_py_version,
        search_dirs=search_dirs,
        download=False,
        app_data=app_data,
        do_periodic_update=False,
        env=env,
    )
    if wheel is None:
        raise RuntimeError("could not find the embedded pip")
    env[str("PYTHONPATH")] = str(wheel.path)
    return env