Add typings for the scripts.compile module

piptools/repositories/base.py

@@ -1,11 +1,19 @@
+import optparse
 from abc import ABCMeta, abstractmethod
 from contextlib import contextmanager
 from typing import Iterator, Optional, Set
 
+from pip._internal.index.package_finder import PackageFinder
+from pip._internal.network.session import PipSession
 from pip._internal.req import InstallRequirement
 
 
 class BaseRepository(metaclass=ABCMeta):
+    DEFAULT_INDEX_URL: str
+    finder: PackageFinder
+    session: PipSession
+    options: optparse.Values
+
     def clear_caches(self) -> None:
         """Should clear any caches used by the implementation."""
 

piptools/repositories/local.py

@@ -1,5 +1,7 @@
 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
@@ -29,7 +31,12 @@ 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: Mapping[str, InstallRequirement],
+        proxied_repository: BaseRepository,
+        reuse_hashes: bool = True,
+    ):
         self._reuse_hashes = reuse_hashes
         self.repository = proxied_repository
         self.existing_pins = existing_pins

piptools/repositories/pypi.py

@@ -6,15 +6,17 @@
 import tempfile
 from contextlib import contextmanager
 from shutil import rmtree
+from typing import Dict, List
 
 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
@@ -51,7 +53,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) -> None:
         # 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.
@@ -77,12 +79,14 @@ 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, List[InstallRequirement]
+        ] = {}
 
         # Setup file paths
         self._build_dir = None
@@ -447,7 +451,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.
@@ -463,7 +467,7 @@ def _setup_logging(self):
         logger = logging.getLogger()
         for handler in logger.handlers:
             if handler.name == "console":  # pragma: no branch
-                handler.stream = log.stream
+                handler.stream = log.stream  # type: ignore[attr-defined]
                 break
         else:  # pragma: no cover
             # There is always a console handler. This warning would be a signal that

piptools/scripts/compile.py

@@ -3,11 +3,10 @@
 import sys
 import tempfile
 import warnings
-from typing import Any
+from typing import Any, List, Optional, Set, Tuple, cast
 
 import click
-from click import Command
-from click.utils import safecall
+from click.utils import LazyFile, safecall
 from pip._internal.commands import create_command
 from pip._internal.req.constructors import install_req_from_line
 from pip._internal.utils.misc import redact_auth_from_url
@@ -18,6 +17,7 @@
 from ..locations import CACHE_DIR
 from ..logging import log
 from ..repositories import LocalRequirementsRepository, PyPIRepository
+from ..repositories.base import BaseRepository
 from ..resolver import Resolver
 from ..utils import UNSAFE_PACKAGES, dedup, is_pinned_requirement, key_from_ireq
 from ..writer import OutputWriter
@@ -36,17 +36,17 @@ def _get_default_option(option_name: str) -> Any:
     return getattr(default_values, option_name)
 
 
-class BaseCommand(Command):
-    _os_args = None
+class BaseCommand(click.Command):
+    _os_args: Set[str]
 
-    def parse_args(self, ctx, args):
+    def parse_args(self, ctx: click.Context, args: List[str]) -> List[str]:
         """
         Override base `parse_args` to store the argument part of `sys.argv`.
         """
         self._os_args = set(args)
         return super().parse_args(ctx, args)
 
-    def has_arg(self, arg_name):
+    def has_arg(self, arg_name: str) -> bool:
         """
         Detect whether a given arg name (including negative counterparts
         to the arg, e.g. --no-arg) is present in the argument part of `sys.argv`.
@@ -216,44 +216,46 @@ def has_arg(self, arg_name):
     show_default=True,
     type=click.Path(file_okay=False, writable=True),
 )
-@click.option("--pip-args", help="Arguments to pass directly to the pip command.")
+@click.option(
+    "--pip-args", "pip_args_str", help="Arguments to pass directly to the pip command."
+)
 @click.option(
     "--emit-index-url/--no-emit-index-url",
     is_flag=True,
     default=True,
     help="Add index URL to generated file",
 )
 def cli(
-    ctx,
-    verbose,
-    quiet,
-    dry_run,
-    pre,
-    rebuild,
-    find_links,
-    index_url,
-    extra_index_url,
-    cert,
-    client_cert,
-    trusted_host,
-    header,
-    index,
-    emit_trusted_host,
-    annotate,
-    upgrade,
-    upgrade_packages,
-    output_file,
-    allow_unsafe,
-    generate_hashes,
-    reuse_hashes,
-    src_files,
-    max_rounds,
-    build_isolation,
-    emit_find_links,
-    cache_dir,
-    pip_args,
-    emit_index_url,
-):
+    ctx: click.Context,
+    verbose: int,
+    quiet: int,
+    dry_run: bool,
+    pre: bool,
+    rebuild: bool,
+    find_links: Tuple[str],
+    index_url: str,
+    extra_index_url: Tuple[str],
+    cert: Optional[str],
+    client_cert: Optional[str],
+    trusted_host: Tuple[str],
+    header: bool,
+    index: bool,
+    emit_trusted_host: bool,
+    annotate: bool,
+    upgrade: bool,
+    upgrade_packages: Tuple[str],
+    output_file: Optional[LazyFile],
+    allow_unsafe: bool,
+    generate_hashes: bool,
+    reuse_hashes: bool,
+    src_files: Tuple[str],
+    max_rounds: int,
+    build_isolation: bool,
+    emit_find_links: bool,
+    cache_dir: str,
+    pip_args_str: Optional[str],
+    emit_index_url: bool,
+) -> None:
     """Compiles requirements.txt from requirements.in specs."""
     log.verbosity = verbose - quiet
 
@@ -292,14 +294,16 @@ def cli(
         output_file = click.open_file(file_name, "w+b", atomic=True, lazy=True)
 
         # Close the file at the end of the context execution
+        assert output_file is not None
         ctx.call_on_close(safecall(output_file.close_intelligently))
 
-    if cli.has_arg("index") and cli.has_arg("emit_index_url"):
+    command = cast(BaseCommand, ctx.command)
+    if command.has_arg("index") and command.has_arg("emit_index_url"):
         raise click.BadParameter(
             "--index/--no-index and --emit-index-url/--no-emit-index-url "
             "are mutually exclusive."
         )
-    elif cli.has_arg("index"):
+    elif command.has_arg("index"):
         warnings.warn(
             "--index and --no-index are deprecated and will be removed "
             "in future versions. Use --emit-index-url/--no-emit-index-url instead.",
@@ -311,7 +315,7 @@ def cli(
     # Setup
     ###
 
-    right_args = shlex.split(pip_args or "")
+    right_args = shlex.split(pip_args_str or "")
     pip_args = []
     for link in find_links:
         pip_args.extend(["-f", link])
@@ -332,6 +336,7 @@ def cli(
         pip_args.append("--no-build-isolation")
     pip_args.extend(right_args)
 
+    repository: BaseRepository
     repository = PyPIRepository(pip_args, cache_dir=cache_dir)
 
     # Parse all constraints coming from --upgrade-package/-P
@@ -385,8 +390,8 @@ def cli(
                 from distutils.core import run_setup
 
                 dist = run_setup(src_file)
-                tmpfile.write("\n".join(dist.install_requires))
-                comes_from = f"{dist.get_name()} ({src_file})"
+                tmpfile.write("\n".join(dist.install_requires))  # type: ignore[attr-defined]
+                comes_from = f"{dist.get_name()} ({src_file})"  # type: ignore[attr-defined]
             else:
                 tmpfile.write(sys.stdin.read())
                 comes_from = "-r -"
@@ -448,10 +453,7 @@ def cli(
             allow_unsafe=allow_unsafe,
         )
         results = resolver.resolve(max_rounds=max_rounds)
-        if generate_hashes:
-            hashes = resolver.resolve_hashes(results)
-        else:
-            hashes = None
+        hashes = resolver.resolve_hashes(results) if generate_hashes else None
     except PipToolsError as e:
         log.error(str(e))
         sys.exit(2)

piptools/utils.py

@@ -315,7 +315,7 @@ def get_compile_command(click_ctx: click.Context) -> str:
             else:
                 if isinstance(val, str) and is_url(val):
                     val = redact_auth_from_url(val)
-                if option.name == "pip_args":
+                if option.name == "pip_args_str":
                     # shlex.quote() would produce functional but noisily quoted results,
                     # e.g. --pip-args='--cache-dir='"'"'/tmp/with spaces'"'"''
                     # Instead, we try to get more legible quoting via repr:

piptools/writer.py

@@ -1,10 +1,11 @@
 import os
 import re
 from itertools import chain
-from typing import BinaryIO, Dict, Iterator, List, Optional, Sequence, Set, Tuple
+from typing import Dict, Iterable, Iterator, List, Optional, Set, Tuple
 
 from click import unstyle
 from click.core import Context
+from click.utils import LazyFile
 from pip._internal.models.format_control import FormatControl
 from pip._internal.req.req_install import InstallRequirement
 from pip._vendor.packaging.markers import Marker
@@ -53,7 +54,7 @@ def _comes_from_as_string(ireq: InstallRequirement) -> str:
 class OutputWriter:
     def __init__(
         self,
-        dst_file: BinaryIO,
+        dst_file: LazyFile,
         click_ctx: Context,
         dry_run: bool,
         emit_header: bool,
@@ -62,8 +63,8 @@ def __init__(
         annotate: bool,
         generate_hashes: bool,
         default_index_url: str,
-        index_urls: Sequence[str],
-        trusted_hosts: Sequence[str],
+        index_urls: Iterable[str],
+        trusted_hosts: Iterable[str],
         format_control: FormatControl,
         allow_unsafe: bool,
         find_links: List[str],
@@ -218,8 +219,8 @@ def write(
         for line in self._iter_lines(results, unsafe_requirements, markers, hashes):
             log.info(line)
             if not self.dry_run:
-                self.dst_file.write(unstyle(line).encode())
-                self.dst_file.write(os.linesep.encode())
+                self.dst_file.write(unstyle(line).encode())  # type: ignore[attr-defined]
+                self.dst_file.write(os.linesep.encode())  # type: ignore[attr-defined]
 
     def _format_requirement(
         self,