Add typing to several areas where straightforward

Refs: #972

piptools/__main__.py

@@ -4,7 +4,7 @@
 
 
 @click.group()
-def cli():
+def cli() -> None:
     pass
 
 

piptools/_compat/pip_compat.py

@@ -1,4 +1,10 @@
+import optparse
+from typing import Iterator, Optional
+
 import pip
+from pip._internal.index.package_finder import PackageFinder
+from pip._internal.network.session import PipSession
+from pip._internal.req import InstallRequirement
 from pip._internal.req import parse_requirements as _parse_requirements
 from pip._internal.req.constructors import install_req_from_parsed_requirement
 from pip._vendor.packaging.version import parse as parse_version
@@ -7,8 +13,13 @@
 
 
 def parse_requirements(
-    filename, session, finder=None, options=None, constraint=False, isolated=False
-):
+    filename: str,
+    session: PipSession,
+    finder: Optional[PackageFinder] = None,
+    options: Optional[optparse.Values] = None,
+    constraint: bool = False,
+    isolated: bool = False,
+) -> Iterator[InstallRequirement]:
     for parsed_req in _parse_requirements(
         filename, session, finder=finder, options=options, constraint=constraint
     ):

piptools/cache.py

@@ -2,16 +2,22 @@
 import os
 import platform
 import sys
+from typing import Dict, List, Optional, Tuple, cast
 
+from pip._internal.req import InstallRequirement
 from pip._vendor.packaging.requirements import Requirement
 
 from .exceptions import PipToolsError
 from .utils import as_tuple, key_from_req, lookup_table
 
+CacheKey = Tuple[str, str]
+CacheLookup = Dict[str, List[str]]
+CacheDict = Dict[str, CacheLookup]
+
 _PEP425_PY_TAGS = {"cpython": "cp", "pypy": "pp", "ironpython": "ip", "jython": "jy"}
 
 
-def _implementation_name():
+def _implementation_name() -> str:
     """
     Similar to PEP 425, however the minor version is separated from the major to
     differentiate "3.10" and "31.0".
@@ -22,10 +28,10 @@ def _implementation_name():
 
 
 class CorruptCacheError(PipToolsError):
-    def __init__(self, path):
+    def __init__(self, path: str):
         self.path = path
 
-    def __str__(self):
+    def __str__(self) -> str:
         lines = [
             "The dependency cache seems to have been corrupted.",
             "Inspect, or delete, the following file:",
@@ -34,7 +40,7 @@ def __str__(self):
         return os.linesep.join(lines)
 
 
-def read_cache_file(cache_file_path):
+def read_cache_file(cache_file_path: str) -> CacheDict:
     with open(cache_file_path) as cache_file:
         try:
             doc = json.load(cache_file)
@@ -44,7 +50,7 @@ def read_cache_file(cache_file_path):
         # Check version and load the contents
         if doc["__format__"] != 1:
             raise ValueError("Unknown cache file format")
-        return doc["dependencies"]
+        return cast(CacheDict, doc["dependencies"])
 
 
 class DependencyCache:
@@ -59,24 +65,27 @@ class DependencyCache:
     Where X.Y indicates the Python version.
     """
 
-    def __init__(self, cache_dir):
+    def __init__(self, cache_dir: str):
         os.makedirs(cache_dir, exist_ok=True)
         cache_filename = f"depcache-{_implementation_name()}.json"
 
         self._cache_file = os.path.join(cache_dir, cache_filename)
-        self._cache = None
+        self._cache: Optional[CacheDict] = None
 
     @property
-    def cache(self):
+    def cache(self) -> CacheDict:
         """
         The dictionary that is the actual in-memory cache.  This property
         lazily loads the cache from disk.
         """
         if self._cache is None:
-            self.read_cache()
+            try:
+                self._cache = read_cache_file(self._cache_file)
+            except FileNotFoundError:
+                self._cache = {}
         return self._cache
 
-    def as_cache_key(self, ireq):
+    def as_cache_key(self, ireq: InstallRequirement) -> CacheKey:
         """
         Given a requirement, return its cache key. This behavior is a little weird
         in order to allow backwards compatibility with cache files. For a requirement
@@ -96,32 +105,25 @@ def as_cache_key(self, ireq):
             extras_string = f"[{','.join(extras)}]"
         return name, f"{version}{extras_string}"
 
-    def read_cache(self):
-        """Reads the cached contents into memory."""
-        try:
-            self._cache = read_cache_file(self._cache_file)
-        except FileNotFoundError:
-            self._cache = {}
-
-    def write_cache(self):
+    def write_cache(self) -> None:
         """Writes the cache to disk as JSON."""
         doc = {"__format__": 1, "dependencies": self._cache}
         with open(self._cache_file, "w") as f:
             json.dump(doc, f, sort_keys=True)
 
-    def clear(self):
+    def clear(self) -> None:
         self._cache = {}
         self.write_cache()
 
-    def __contains__(self, ireq):
+    def __contains__(self, ireq: InstallRequirement) -> bool:
         pkgname, pkgversion_and_extras = self.as_cache_key(ireq)
         return pkgversion_and_extras in self.cache.get(pkgname, {})
 
-    def __getitem__(self, ireq):
+    def __getitem__(self, ireq: InstallRequirement) -> List[str]:
         pkgname, pkgversion_and_extras = self.as_cache_key(ireq)
         return self.cache[pkgname][pkgversion_and_extras]
 
-    def __setitem__(self, ireq, values):
+    def __setitem__(self, ireq: InstallRequirement, values: List[str]) -> None:
         pkgname, pkgversion_and_extras = self.as_cache_key(ireq)
         self.cache.setdefault(pkgname, {})
         self.cache[pkgname][pkgversion_and_extras] = values

piptools/exceptions.py

@@ -1,3 +1,8 @@
+from typing import Iterable
+
+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.misc import redact_auth_from_url
 
 
@@ -6,12 +11,17 @@ class PipToolsError(Exception):
 
 
 class NoCandidateFound(PipToolsError):
-    def __init__(self, ireq, candidates_tried, finder):
+    def __init__(
+        self,
+        ireq: InstallRequirement,
+        candidates_tried: Iterable[InstallationCandidate],
+        finder: PackageFinder,
+    ) -> None:
         self.ireq = ireq
         self.candidates_tried = candidates_tried
         self.finder = finder
 
-    def __str__(self):
+    def __str__(self) -> str:
         versions = []
         pre_versions = []
 
@@ -57,10 +67,10 @@ def __str__(self):
 
 
 class IncompatibleRequirements(PipToolsError):
-    def __init__(self, ireq_a, ireq_b):
+    def __init__(self, ireq_a: InstallRequirement, ireq_b: InstallRequirement) -> None:
         self.ireq_a = ireq_a
         self.ireq_b = ireq_b
 
-    def __str__(self):
+    def __str__(self) -> str:
         message = "Incompatible requirements found: {} and {}"
         return message.format(self.ireq_a, self.ireq_b)

piptools/logging.py

@@ -1,7 +1,7 @@
 import contextlib
 import logging
 import sys
-from typing import Any
+from typing import Any, Iterator
 
 import click
 
@@ -23,30 +23,30 @@ def log(self, message: str, *args: Any, **kwargs: Any) -> None:
         prefix = " " * self.current_indent
         click.secho(prefix + message, *args, **kwargs)
 
-    def debug(self, *args, **kwargs):
+    def debug(self, message: str, *args: Any, **kwargs: Any) -> None:
         if self.verbosity >= 1:
-            self.log(*args, **kwargs)
+            self.log(message, *args, **kwargs)
 
-    def info(self, *args: Any, **kwargs: Any) -> None:
+    def info(self, message: str, *args: Any, **kwargs: Any) -> None:
         if self.verbosity >= 0:
-            self.log(*args, **kwargs)
+            self.log(message, *args, **kwargs)
 
-    def warning(self, *args: Any, **kwargs: Any) -> None:
+    def warning(self, message: str, *args: Any, **kwargs: Any) -> None:
         kwargs.setdefault("fg", "yellow")
-        self.log(*args, **kwargs)
+        self.log(message, *args, **kwargs)
 
-    def error(self, *args, **kwargs):
+    def error(self, message: str, *args: Any, **kwargs: Any) -> None:
         kwargs.setdefault("fg", "red")
-        self.log(*args, **kwargs)
+        self.log(message, *args, **kwargs)
 
-    def _indent(self):
+    def _indent(self) -> None:
         self.current_indent += self._indent_width
 
-    def _dedent(self):
+    def _dedent(self) -> None:
         self.current_indent -= self._indent_width
 
     @contextlib.contextmanager
-    def indentation(self):
+    def indentation(self) -> Iterator[None]:
         """
         Increase indentation.
         """

piptools/repositories/base.py

@@ -1,33 +1,37 @@
 from abc import ABCMeta, abstractmethod
 from contextlib import contextmanager
+from typing import Iterator, Set
+
+from pip._internal.req import InstallRequirement
+from pip._vendor.packaging.version import Version
 
 
 class BaseRepository(metaclass=ABCMeta):
-    def clear_caches(self):
+    def clear_caches(self) -> None:
         """Should clear any caches used by the implementation."""
 
     @abstractmethod
     @contextmanager
-    def freshen_build_caches(self):
+    def freshen_build_caches(self) -> Iterator[None]:
         """Should start with fresh build/source caches."""
 
     @abstractmethod
-    def find_best_match(self, ireq):
+    def find_best_match(self, ireq: InstallRequirement) -> Version:
         """
         Return a Version object that indicates the best match for the given
         InstallRequirement according to the repository.
         """
 
     @abstractmethod
-    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).
         They indicate the secondary dependencies for the given requirement.
         """
 
     @abstractmethod
-    def get_hashes(self, ireq):
+    def get_hashes(self, ireq: InstallRequirement) -> Set[str]:
         """
         Given a pinned InstallRequire, returns a set of hashes that represent
         all of the files for a given requirement. It is not acceptable for an
@@ -36,13 +40,15 @@ def get_hashes(self, ireq):
 
     @abstractmethod
     @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.
         """
 
     @abstractmethod
-    def copy_ireq_dependencies(self, source, dest):
+    def copy_ireq_dependencies(
+        self, source: InstallRequirement, dest: InstallRequirement
+    ) -> None:
         """
         Notifies the repository that `dest` is a copy of `source`, and so it
         has the same dependencies. Otherwise, once we prepare an ireq to assign

piptools/resolver.py

@@ -3,6 +3,7 @@
 from itertools import chain, count, groupby
 
 import click
+from pip._internal.req import InstallRequirement
 from pip._internal.req.constructors import install_req_from_line
 from pip._internal.req.req_tracker import update_env_context_manager
 
@@ -25,23 +26,25 @@ class RequirementSummary:
     Summary of a requirement's properties for comparison purposes.
     """
 
-    def __init__(self, ireq):
+    def __init__(self, ireq: InstallRequirement):
         self.req = ireq.req
         self.key = key_from_ireq(ireq)
         self.extras = frozenset(ireq.extras)
         self.specifier = ireq.specifier
 
-    def __eq__(self, other):
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, RequirementSummary):
+            return NotImplemented
         return (
             self.key == other.key
             and self.specifier == other.specifier
             and self.extras == other.extras
         )
 
-    def __hash__(self):
+    def __hash__(self) -> int:
         return hash((self.key, self.specifier, self.extras))
 
-    def __str__(self):
+    def __str__(self) -> str:
         return repr((self.key, str(self.specifier), sorted(self.extras)))