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