local.py

from contextlib import contextmanager
from typing import Mapping

from pip._internal.req import InstallRequirement
from pip._internal.utils.hashes import FAVORITE_HASH

from piptools.utils import as_tuple, key_from_ireq, make_install_requirement

from .base import BaseRepository


def ireq_satisfied_by_existing_pin(ireq, existing_pin):
    """
    Return True if the given InstallationRequirement is satisfied by the
    previously encountered version pin.
    """
    version = next(iter(existing_pin.req.specifier)).version
    return ireq.req.specifier.contains(
        version, prereleases=existing_pin.req.specifier.prereleases
    )


class LocalRequirementsRepository(BaseRepository):
    """
    The LocalRequirementsRepository proxied the _real_ repository by first
    checking if a requirement can be satisfied by existing pins (i.e. the
    result of a previous compile step).

    In effect, if a requirement can be satisfied with a version pinned in the
    requirements file, we prefer that version over the best match found in
    PyPI.  This keeps updates to the requirements.txt down to a minimum.
    """

    def __init__(
        self,
        existing_pins: Mapping[str, InstallRequirement],
        proxied_repository: BaseRepository,
        reuse_hashes: bool = True,
    ):
        self._reuse_hashes = reuse_hashes
        self.repository = proxied_repository
        self.existing_pins = existing_pins

    @property
    def options(self):
        return self.repository.options

    @property
    def finder(self):
        return self.repository.finder

    @property
    def session(self):
        return self.repository.session

    @property
    def DEFAULT_INDEX_URL(self):
        return self.repository.DEFAULT_INDEX_URL

    def clear_caches(self):
        self.repository.clear_caches()

    @contextmanager
    def freshen_build_caches(self):
        with self.repository.freshen_build_caches():
            yield

    def find_best_match(self, ireq, prereleases=None):
        key = key_from_ireq(ireq)
        existing_pin = self.existing_pins.get(key)
        if existing_pin and ireq_satisfied_by_existing_pin(ireq, existing_pin):
            project, version, _ = as_tuple(existing_pin)
            return make_install_requirement(
                project, version, ireq.extras, constraint=ireq.constraint
            )
        else:
            return self.repository.find_best_match(ireq, prereleases)

    def get_dependencies(self, ireq):
        return self.repository.get_dependencies(ireq)

    def get_hashes(self, ireq):
        existing_pin = self._reuse_hashes and self.existing_pins.get(
            key_from_ireq(ireq)
        )
        if existing_pin and ireq_satisfied_by_existing_pin(ireq, existing_pin):
            hashes = existing_pin.hash_options
            hexdigests = hashes.get(FAVORITE_HASH)
            if hexdigests:
                return {
                    ":".join([FAVORITE_HASH, hexdigest]) for hexdigest in hexdigests
                }
        return self.repository.get_hashes(ireq)

    @contextmanager
    def allow_all_wheels(self):
        with self.repository.allow_all_wheels():
            yield

    def copy_ireq_dependencies(self, source, dest):
        self.repository.copy_ireq_dependencies(source, dest)