diff --git a/piptools/_compat/contextlib.py b/piptools/_compat/contextlib.py index dd30ca71f..8f0c5c863 100644 --- a/piptools/_compat/contextlib.py +++ b/piptools/_compat/contextlib.py @@ -16,10 +16,10 @@ class nullcontext: TODO: replace with `contextlib.nullcontext()` after Python 3.6 being dropped """ - def __init__(self, enter_result: Optional[_T] = None) -> None: + def __init__(self, enter_result: _T) -> None: self.enter_result = enter_result - def __enter__(self) -> Optional[_T]: + def __enter__(self) -> _T: return self.enter_result def __exit__( diff --git a/piptools/cache.py b/piptools/cache.py index dd24ff5b1..da325b27d 100644 --- a/piptools/cache.py +++ b/piptools/cache.py @@ -8,7 +8,7 @@ from pip._vendor.packaging.requirements import Requirement from .exceptions import PipToolsError -from .utils import as_tuple, key_from_req, lookup_table +from .utils import as_tuple, key_from_req, lookup_table_from_tuples CacheKey = Tuple[str, str] CacheLookup = Dict[str, List[str]] @@ -166,7 +166,7 @@ def _reverse_dependencies( """ # First, collect all the dependencies into a sequence of (parent, child) # tuples, like [('flake8', 'pep8'), ('flake8', 'mccabe'), ...] - return lookup_table( + return lookup_table_from_tuples( (key_from_req(Requirement(dep_name)), name) for name, version_and_extras in cache_keys for dep_name in self.cache[name][version_and_extras] diff --git a/piptools/repositories/local.py b/piptools/repositories/local.py index 4cb04178b..e038b1358 100644 --- a/piptools/repositories/local.py +++ b/piptools/repositories/local.py @@ -1,21 +1,31 @@ +import optparse from contextlib import contextmanager +from typing import Dict, Iterator, List, Optional, Set, cast +from pip._internal.index.package_finder import PackageFinder +from pip._internal.models.candidate import InstallationCandidate +from pip._internal.req import InstallRequirement from pip._internal.utils.hashes import FAVORITE_HASH +from pip._vendor.requests import Session from piptools.utils import as_tuple, key_from_ireq, make_install_requirement from .base import BaseRepository +from .pypi import PyPIRepository -def ireq_satisfied_by_existing_pin(ireq, existing_pin): +def ireq_satisfied_by_existing_pin( + ireq: InstallRequirement, existing_pin: InstallationCandidate +) -> bool: """ 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( + result = ireq.req.specifier.contains( version, prereleases=existing_pin.req.specifier.prereleases ) + return cast(bool, result) class LocalRequirementsRepository(BaseRepository): @@ -29,36 +39,43 @@ class LocalRequirementsRepository(BaseRepository): PyPI. This keeps updates to the requirements.txt down to a minimum. """ - def __init__(self, existing_pins, proxied_repository, reuse_hashes=True): + def __init__( + self, + existing_pins: Dict[str, InstallationCandidate], + proxied_repository: PyPIRepository, + 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 + def options(self) -> List[optparse.Option]: + return cast(List[optparse.Option], self.repository.options) @property - def finder(self): + def finder(self) -> PackageFinder: return self.repository.finder @property - def session(self): + def session(self) -> Session: return self.repository.session @property - def DEFAULT_INDEX_URL(self): - return self.repository.DEFAULT_INDEX_URL + def DEFAULT_INDEX_URL(self) -> str: + return cast(str, self.repository.DEFAULT_INDEX_URL) - def clear_caches(self): + def clear_caches(self) -> None: self.repository.clear_caches() @contextmanager - def freshen_build_caches(self): + def freshen_build_caches(self) -> Iterator[None]: with self.repository.freshen_build_caches(): yield - def find_best_match(self, ireq, prereleases=None): + def find_best_match( + self, ireq: InstallRequirement, prereleases: Optional[bool] = None + ) -> InstallationCandidate: key = key_from_ireq(ireq) existing_pin = self.existing_pins.get(key) if existing_pin and ireq_satisfied_by_existing_pin(ireq, existing_pin): @@ -67,10 +84,10 @@ def find_best_match(self, ireq, prereleases=None): else: return self.repository.find_best_match(ireq, prereleases) - def get_dependencies(self, ireq): + def get_dependencies(self, ireq: InstallRequirement) -> Set[InstallRequirement]: return self.repository.get_dependencies(ireq) - def get_hashes(self, ireq): + def get_hashes(self, ireq: InstallRequirement) -> Set[str]: existing_pin = self._reuse_hashes and self.existing_pins.get( key_from_ireq(ireq) ) @@ -84,9 +101,11 @@ def get_hashes(self, ireq): return self.repository.get_hashes(ireq) @contextmanager - def allow_all_wheels(self): + def allow_all_wheels(self) -> Iterator[None]: with self.repository.allow_all_wheels(): yield - def copy_ireq_dependencies(self, source, dest): + def copy_ireq_dependencies( + self, source: InstallRequirement, dest: InstallRequirement + ) -> None: self.repository.copy_ireq_dependencies(source, dest) diff --git a/piptools/repositories/pypi.py b/piptools/repositories/pypi.py index 78d36b27b..00b923117 100644 --- a/piptools/repositories/pypi.py +++ b/piptools/repositories/pypi.py @@ -6,22 +6,26 @@ import tempfile from contextlib import contextmanager from shutil import rmtree +from typing import Any, ContextManager, Dict, Iterator, List, Optional, Set, cast from click import progressbar from pip._internal.cache import WheelCache from pip._internal.cli.progress_bars import BAR_TYPES from pip._internal.commands import create_command +from pip._internal.models.candidate import InstallationCandidate from pip._internal.models.index import PackageIndex, PyPI from pip._internal.models.link import Link from pip._internal.models.wheel import Wheel -from pip._internal.req import RequirementSet +from pip._internal.req import InstallRequirement, RequirementSet from pip._internal.req.req_tracker import get_requirement_tracker from pip._internal.utils.hashes import FAVORITE_HASH from pip._internal.utils.logging import indent_log, setup_logging from pip._internal.utils.misc import normalize_path from pip._internal.utils.temp_dir import TempDirectory, global_tempdir_manager from pip._internal.utils.urls import path_to_url, url_to_path -from pip._vendor.requests import RequestException +from pip._vendor.packaging.tags import Tag +from pip._vendor.packaging.version import _BaseVersion +from pip._vendor.requests import RequestException, Session from .._compat import contextlib from ..exceptions import NoCandidateFound @@ -50,7 +54,7 @@ class PyPIRepository(BaseRepository): changed/configured on the Finder. """ - def __init__(self, pip_args, cache_dir): + def __init__(self, pip_args: List[str], cache_dir: str): # Use pip's parser for pip.conf management and defaults. # General options (find_links, index_url, extra_index_url, trusted_host, # and pre) are deferred to pip. @@ -72,23 +76,23 @@ def __init__(self, pip_args, cache_dir): # stores project_name => InstallationCandidate mappings for all # versions reported by PyPI, so we only have to ask once for each # project - self._available_candidates_cache = {} + self._available_candidates_cache: Dict[str, List[InstallationCandidate]] = {} # stores InstallRequirement => list(InstallRequirement) mappings # of all secondary dependencies for the given requirement, so we # only have to go to disk once for each requirement - self._dependencies_cache = {} + self._dependencies_cache: Dict[InstallRequirement, Set[InstallRequirement]] = {} # Setup file paths - self._build_dir = None - self._source_dir = None + self._build_dir: Optional[tempfile.TemporaryDirectory[str]] = None + self._source_dir: Optional[tempfile.TemporaryDirectory[str]] = None self._cache_dir = normalize_path(str(cache_dir)) self._download_dir = os.path.join(self._cache_dir, "pkgs") self._setup_logging() @contextmanager - def freshen_build_caches(self): + def freshen_build_caches(self) -> Iterator[None]: """ Start with fresh build/source caches. Will remove any old build caches from disk automatically. @@ -104,23 +108,25 @@ def freshen_build_caches(self): self._source_dir = None @property - def build_dir(self): + def build_dir(self) -> Optional[str]: return self._build_dir.name if self._build_dir else None @property - def source_dir(self): + def source_dir(self) -> Optional[str]: return self._source_dir.name if self._source_dir else None - def clear_caches(self): + def clear_caches(self) -> None: rmtree(self._download_dir, ignore_errors=True) - def find_all_candidates(self, req_name): + def find_all_candidates(self, req_name: str) -> List[InstallationCandidate]: if req_name not in self._available_candidates_cache: candidates = self.finder.find_all_candidates(req_name) self._available_candidates_cache[req_name] = candidates return self._available_candidates_cache[req_name] - def find_best_match(self, ireq, prereleases=None): + def find_best_match( + self, ireq: InstallRequirement, prereleases: Optional[bool] = None + ) -> InstallRequirement: """ Returns a pinned InstallRequirement object that indicates the best match for the given InstallRequirement according to the external repository. @@ -129,7 +135,7 @@ def find_best_match(self, ireq, prereleases=None): return ireq # return itself as the best match all_candidates = self.find_all_candidates(ireq.name) - candidates_by_version = lookup_table(all_candidates, key=lambda c: c.version) + candidates_by_version = lookup_table(all_candidates, key=candidate_version) matching_versions = ireq.specifier.filter( (candidate.version for candidate in all_candidates), prereleases=prereleases ) @@ -153,7 +159,12 @@ def find_best_match(self, ireq, prereleases=None): ireq, ) - def resolve_reqs(self, download_dir, ireq, wheel_cache): + def resolve_reqs( + self, + download_dir: Optional[str], + ireq: InstallRequirement, + wheel_cache: WheelCache, + ) -> Set[InstallationCandidate]: with get_requirement_tracker() as req_tracker, TempDirectory( kind="resolver" ) as temp_dir, indent_log(): @@ -191,7 +202,7 @@ def resolve_reqs(self, download_dir, ireq, wheel_cache): return set(results) - def get_dependencies(self, ireq): + def get_dependencies(self, ireq: InstallRequirement) -> Set[InstallRequirement]: """ Given a pinned, URL, or editable InstallRequirement, returns a set of dependencies (also InstallRequirements, but not necessarily pinned). @@ -226,14 +237,16 @@ def get_dependencies(self, ireq): return self._dependencies_cache[ireq] - def copy_ireq_dependencies(self, source, dest): + def copy_ireq_dependencies( + self, source: InstallRequirement, dest: InstallRequirement + ) -> None: try: self._dependencies_cache[dest] = self._dependencies_cache[source] except KeyError: # `source` may not be in cache yet. pass - def _get_project(self, ireq): + def _get_project(self, ireq: InstallRequirement) -> Any: """ Return a dict of a project info from PyPI JSON API for a given InstallRequirement. Return None on HTTP/JSON error or if a package @@ -266,7 +279,7 @@ def _get_project(self, ireq): return data return None - def _get_download_path(self, ireq): + def _get_download_path(self, ireq: InstallRequirement) -> str: """ Determine the download dir location in a way which avoids name collisions. @@ -275,12 +288,13 @@ def _get_download_path(self, ireq): salt = hashlib.sha224(ireq.link.url_without_fragment.encode()).hexdigest() # Nest directories to avoid running out of top level dirs on some FS # (see pypi _get_cache_path_parts, which inspired this) - salt = [salt[:2], salt[2:4], salt[4:6], salt[6:]] - return os.path.join(self._download_dir, *salt) + return os.path.join( + self._download_dir, salt[:2], salt[2:4], salt[4:6], salt[6:] + ) else: return self._download_dir - def get_hashes(self, ireq): + def get_hashes(self, ireq: InstallRequirement) -> Set[str]: """ Given an InstallRequirement, return a set of hashes that represent all of the files for a given requirement. Unhashable requirements return an @@ -320,7 +334,7 @@ def get_hashes(self, ireq): return hashes - def _get_hashes_from_pypi(self, ireq): + def _get_hashes_from_pypi(self, ireq: InstallRequirement) -> Optional[Set[str]]: """ Return a set of hashes from PyPI JSON API for a given InstallRequirement. Return None if fetching data is failed or missing digests. @@ -349,7 +363,7 @@ def _get_hashes_from_pypi(self, ireq): return hashes - def _get_hashes_from_files(self, ireq): + def _get_hashes_from_files(self, ireq: InstallRequirement) -> Set[str]: """ Return a set of hashes for all release files of a given InstallRequirement. """ @@ -357,7 +371,7 @@ def _get_hashes_from_files(self, ireq): # pin, these will represent all of the files that could possibly # satisfy this constraint. all_candidates = self.find_all_candidates(ireq.name) - candidates_by_version = lookup_table(all_candidates, key=lambda c: c.version) + candidates_by_version = lookup_table(all_candidates, key=candidate_version) matching_versions = list( ireq.specifier.filter(candidate.version for candidate in all_candidates) ) @@ -367,14 +381,15 @@ def _get_hashes_from_files(self, ireq): self._get_file_hash(candidate.link) for candidate in matching_candidates } - def _get_file_hash(self, link): + def _get_file_hash(self, link: Link) -> str: log.debug(f"Hashing {link.show_url}") h = hashlib.new(FAVORITE_HASH) with open_local_or_remote_file(link, self.session) as f: # Chunks to iterate - chunks = iter(lambda: f.stream.read(FILE_CHUNK_SIZE), b"") + chunks = iter(lambda: cast(bytes, f.stream.read(FILE_CHUNK_SIZE)), b"") # Choose a context manager depending on verbosity + context_manager: ContextManager[Iterator[bytes]] if log.verbosity >= 1: iter_length = f.size / FILE_CHUNK_SIZE if f.size else None bar_template = f"{' ' * log.current_indent} |%(bar)s| %(info)s" @@ -397,7 +412,7 @@ def _get_file_hash(self, link): return ":".join([FAVORITE_HASH, h.hexdigest()]) @contextmanager - def allow_all_wheels(self): + def allow_all_wheels(self) -> Iterator[None]: """ Monkey patches pip.Wheel to allow wheels from all platforms and Python versions. @@ -405,11 +420,11 @@ def allow_all_wheels(self): the previous non-patched calls will interfere. """ - def _wheel_supported(self, tags=None): + def _wheel_supported(self: Wheel, tags: List[Tag]) -> bool: # Ignore current platform. Support everything. return True - def _wheel_support_index_min(self, tags=None): + def _wheel_support_index_min(self: Wheel, tags: List[Tag]) -> int: # All wheels are equal priority for sorting. return 0 @@ -428,7 +443,7 @@ def _wheel_support_index_min(self, tags=None): Wheel.support_index_min = original_support_index_min self._available_candidates_cache = original_cache - def _setup_logging(self): + def _setup_logging(self) -> None: """ Setup pip's logger. Ensure pip is verbose same as pip-tools and sync pip's log stream with LogContext.stream. @@ -444,6 +459,7 @@ def _setup_logging(self): logger = logging.getLogger() for handler in logger.handlers: if handler.name == "console": # pragma: no branch + assert isinstance(handler, logging.StreamHandler) handler.stream = log.stream break else: # pragma: no cover @@ -458,7 +474,7 @@ def _setup_logging(self): @contextmanager -def open_local_or_remote_file(link, session): +def open_local_or_remote_file(link: Link, session: Session) -> Iterator[FileStream]: """ Open local or remote file for reading. @@ -484,6 +500,7 @@ def open_local_or_remote_file(link, session): response = session.get(url, headers=headers, stream=True) # Content length must be int or None + content_length: Optional[int] try: content_length = int(response.headers["content-length"]) except (ValueError, KeyError, TypeError): @@ -493,3 +510,7 @@ def open_local_or_remote_file(link, session): yield FileStream(stream=response.raw, size=content_length) finally: response.close() + + +def candidate_version(candidate: InstallationCandidate) -> _BaseVersion: + return candidate.version diff --git a/piptools/utils.py b/piptools/utils.py index ed33aa602..8deb39321 100644 --- a/piptools/utils.py +++ b/piptools/utils.py @@ -184,33 +184,25 @@ def flat_map( return itertools.chain.from_iterable(map(fn, collection)) -def lookup_table( - values: Iterable[Union[_VT, Tuple[_KT, _VT]]], - key: Optional[Callable[[_VT], _KT]] = None, -) -> Dict[_KT, Set[_VT]]: +def lookup_table_from_tuples(values: Iterable[Tuple[_KT, _VT]]) -> Dict[_KT, Set[_VT]]: """ Builds a dict-based lookup table (index) elegantly. """ - values, values_to_validate = itertools.tee(values) - if key is None and any(not isinstance(v, tuple) for v in values_to_validate): - raise ValueError( - "The `key` function must be specified when the `values` are not empty." - ) - - def keyval(v: Union[_VT, Tuple[_KT, _VT]]) -> Tuple[_KT, _VT]: - if isinstance(v, tuple): - return v[0], v[1] - - assert key is not None, "key function must be specified" - return key(v), v - lut: Dict[_KT, Set[_VT]] = collections.defaultdict(set) - for value in values: - k, v = keyval(value) + for k, v in values: lut[k].add(v) return dict(lut) +def lookup_table( + values: Iterable[_VT], key: Callable[[_VT], _KT] +) -> Dict[_KT, Set[_VT]]: + """ + Builds a dict-based lookup table (index) elegantly. + """ + return lookup_table_from_tuples((key(v), v) for v in values) + + def dedup(iterable: Iterable[_T]) -> Iterable[_T]: """Deduplicate an iterable object like iter(set(iterable)) but order-preserved. diff --git a/tests/test_utils.py b/tests/test_utils.py index c975fe55d..d6f245c3d 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -17,6 +17,7 @@ is_pinned_requirement, is_url_requirement, lookup_table, + lookup_table_from_tuples, ) @@ -337,58 +338,32 @@ def test_get_compile_command_sort_args(tmpdir_cwd): @pytest.mark.parametrize( - ("values", "key"), + "tuples", ( - pytest.param( - ("foo", "bar", "baz", "qux", "quux"), - operator.itemgetter(0), - id="with key function", - ), - pytest.param( - (("f", "foo"), ("b", "bar"), ("b", "baz"), ("q", "qux"), ("q", "quux")), - None, - id="without key function", - ), - pytest.param( - iter(("foo", "bar", "baz", "qux", "quux")), - operator.itemgetter(0), - id="values as iterator with key function", - ), - pytest.param( - iter( - (("f", "foo"), ("b", "bar"), ("b", "baz"), ("q", "qux"), ("q", "quux")) - ), - None, - id="values as iterator without key function", - ), + (("f", "foo"), ("b", "bar"), ("b", "baz"), ("q", "qux"), ("q", "quux")), + iter((("f", "foo"), ("b", "bar"), ("b", "baz"), ("q", "qux"), ("q", "quux"))), ), ) -def test_lookup_table(values, key): +def test_lookup_table_from_tuples(tuples): expected = {"b": {"bar", "baz"}, "f": {"foo"}, "q": {"quux", "qux"}} - assert lookup_table(values, key) == expected + assert lookup_table_from_tuples(tuples) == expected @pytest.mark.parametrize( - "values", + ("values", "key"), ( - pytest.param(("foo", "bar", "baz"), id="values are not tuples"), - pytest.param((("f", "foo"), "b"), id="one of the values is not a tuple"), + (("foo", "bar", "baz", "qux", "quux"), operator.itemgetter(0)), + (iter(("foo", "bar", "baz", "qux", "quux")), operator.itemgetter(0)), ), ) -def test_lookup_table_requires_key(values): - with pytest.raises( - ValueError, - match=r"^The `key` function must be specified when the `values` are not empty\.$", - ): - lookup_table(values) +def test_lookup_table(values, key): + expected = {"b": {"bar", "baz"}, "f": {"foo"}, "q": {"quux", "qux"}} + assert lookup_table(values, key) == expected -@pytest.mark.parametrize( - "key", - ( - pytest.param(lambda x: x, id="with key"), - pytest.param(None, id="without key"), - ), -) -def test_lookup_table_with_empty_values(key): - assert lookup_table((), key) == {} +def test_lookup_table_from_tuples_with_empty_values(): + assert lookup_table_from_tuples(()) == {} + + +def test_lookup_table_with_empty_values(): + assert lookup_table((), operator.itemgetter(0)) == {}