1import optparse 2from contextlib import contextmanager 3from typing import Iterator, Mapping, Optional, Set, cast 4 5from pip._internal.index.package_finder import PackageFinder 6from pip._internal.models.candidate import InstallationCandidate 7from pip._internal.req import InstallRequirement 8from pip._internal.utils.hashes import FAVORITE_HASH 9from pip._vendor.requests import Session 10 11from piptools.utils import as_tuple, key_from_ireq, make_install_requirement 12 13from .base import BaseRepository 14from .pypi import PyPIRepository 15 16 17def ireq_satisfied_by_existing_pin( 18 ireq: InstallRequirement, existing_pin: InstallationCandidate 19) -> bool: 20 """ 21 Return True if the given InstallationRequirement is satisfied by the 22 previously encountered version pin. 23 """ 24 version = next(iter(existing_pin.req.specifier)).version 25 result = ireq.req.specifier.contains( 26 version, prereleases=existing_pin.req.specifier.prereleases 27 ) 28 return cast(bool, result) 29 30 31class LocalRequirementsRepository(BaseRepository): 32 """ 33 The LocalRequirementsRepository proxied the _real_ repository by first 34 checking if a requirement can be satisfied by existing pins (i.e. the 35 result of a previous compile step). 36 37 In effect, if a requirement can be satisfied with a version pinned in the 38 requirements file, we prefer that version over the best match found in 39 PyPI. This keeps updates to the requirements.txt down to a minimum. 40 """ 41 42 def __init__( 43 self, 44 existing_pins: Mapping[str, InstallationCandidate], 45 proxied_repository: PyPIRepository, 46 reuse_hashes: bool = True, 47 ): 48 self._reuse_hashes = reuse_hashes 49 self.repository = proxied_repository 50 self.existing_pins = existing_pins 51 52 @property 53 def options(self) -> optparse.Values: 54 return self.repository.options 55 56 @property 57 def finder(self) -> PackageFinder: 58 return self.repository.finder 59 60 @property 61 def session(self) -> Session: 62 return self.repository.session 63 64 def clear_caches(self) -> None: 65 self.repository.clear_caches() 66 67 def find_best_match( 68 self, ireq: InstallRequirement, prereleases: Optional[bool] = None 69 ) -> InstallationCandidate: 70 key = key_from_ireq(ireq) 71 existing_pin = self.existing_pins.get(key) 72 if existing_pin and ireq_satisfied_by_existing_pin(ireq, existing_pin): 73 project, version, _ = as_tuple(existing_pin) 74 return make_install_requirement(project, version, ireq) 75 else: 76 return self.repository.find_best_match(ireq, prereleases) 77 78 def get_dependencies(self, ireq: InstallRequirement) -> Set[InstallRequirement]: 79 return self.repository.get_dependencies(ireq) 80 81 def get_hashes(self, ireq: InstallRequirement) -> Set[str]: 82 existing_pin = self._reuse_hashes and self.existing_pins.get( 83 key_from_ireq(ireq) 84 ) 85 if existing_pin and ireq_satisfied_by_existing_pin(ireq, existing_pin): 86 hashes = existing_pin.hash_options 87 hexdigests = hashes.get(FAVORITE_HASH) 88 if hexdigests: 89 return { 90 ":".join([FAVORITE_HASH, hexdigest]) for hexdigest in hexdigests 91 } 92 return self.repository.get_hashes(ireq) 93 94 @contextmanager 95 def allow_all_wheels(self) -> Iterator[None]: 96 with self.repository.allow_all_wheels(): 97 yield 98