Merge pull request #8265 from deveshks/add-mypy-download-freeze-search

Complete type annotations in "pip._internal.commands.{freeze,download,search}"

src/pip/_internal/commands/download.py

@@ -1,6 +1,3 @@
-# The following comment should be removed at some point in the future.
-# mypy: disallow-untyped-defs=False
-
 from __future__ import absolute_import
 
 import logging
@@ -9,9 +6,15 @@ import os
 from pip._internal.cli import cmdoptions
 from pip._internal.cli.cmdoptions import make_target_python
 from pip._internal.cli.req_command import RequirementCommand, with_cleanup
+from pip._internal.cli.status_codes import SUCCESS
 from pip._internal.req.req_tracker import get_requirement_tracker
 from pip._internal.utils.misc import ensure_dir, normalize_path, write_output
 from pip._internal.utils.temp_dir import TempDirectory
+from pip._internal.utils.typing import MYPY_CHECK_RUNNING
+
+if MYPY_CHECK_RUNNING:
+    from optparse import Values
+    from typing import Any, List
 
 logger = logging.getLogger(__name__)
 
@@ -37,6 +40,7 @@ class DownloadCommand(RequirementCommand):
       %prog [options] <archive url/path> ..."""
 
     def __init__(self, *args, **kw):
+        # type: (*Any, **Any) -> None
         super(DownloadCommand, self).__init__(*args, **kw)
 
         cmd_opts = self.cmd_opts
@@ -77,6 +81,8 @@ class DownloadCommand(RequirementCommand):
 
     @with_cleanup
     def run(self, options, args):
+        # type: (Values, List[str]) -> int
+
         options.ignore_installed = True
         # editable doesn't really make sense for `pip download`, but the bowels
         # of the RequirementSet code require that property.
@@ -132,11 +138,10 @@ class DownloadCommand(RequirementCommand):
             reqs, check_supported_wheels=True
         )
 
-        downloaded = ' '.join([
+        downloaded = ' '.join([req.name  # type: ignore
-            req.name for req in requirement_set.requirements.values()
+                               for req in requirement_set.requirements.values()
-            if req.successfully_downloaded
+                               if req.successfully_downloaded])
-        ])
         if downloaded:
             write_output('Successfully downloaded %s', downloaded)
 
-        return requirement_set
+        return SUCCESS

src/pip/_internal/commands/freeze.py

@@ -1,6 +1,3 @@
-# The following comment should be removed at some point in the future.
-# mypy: disallow-untyped-defs=False
-
 from __future__ import absolute_import
 
 import sys
@@ -8,12 +5,18 @@ import sys
 from pip._internal.cache import WheelCache
 from pip._internal.cli import cmdoptions
 from pip._internal.cli.base_command import Command
+from pip._internal.cli.status_codes import SUCCESS
 from pip._internal.models.format_control import FormatControl
 from pip._internal.operations.freeze import freeze
 from pip._internal.utils.compat import stdlib_pkgs
+from pip._internal.utils.typing import MYPY_CHECK_RUNNING
 
 DEV_PKGS = {'pip', 'setuptools', 'distribute', 'wheel'}
 
+if MYPY_CHECK_RUNNING:
+    from optparse import Values
+    from typing import Any, List
+
 
 class FreezeCommand(Command):
     """
@@ -27,6 +30,7 @@ class FreezeCommand(Command):
     log_streams = ("ext://sys.stderr", "ext://sys.stderr")
 
     def __init__(self, *args, **kw):
+        # type: (*Any, **Any) -> None
         super(FreezeCommand, self).__init__(*args, **kw)
 
         self.cmd_opts.add_option(
@@ -75,6 +79,7 @@ class FreezeCommand(Command):
         self.parser.insert_option_group(0, self.cmd_opts)
 
     def run(self, options, args):
+        # type: (Values, List[str]) -> int
         format_control = FormatControl(set(), set())
         wheel_cache = WheelCache(options.cache_dir, format_control)
         skip = set(stdlib_pkgs)
@@ -97,3 +102,4 @@ class FreezeCommand(Command):
 
         for line in freeze(**freeze_kwargs):
             sys.stdout.write(line + '\n')
+        return SUCCESS

src/pip/_internal/commands/search.py

@@ -1,6 +1,3 @@
-# The following comment should be removed at some point in the future.
-# mypy: disallow-untyped-defs=False
-
 from __future__ import absolute_import
 
 import logging
@@ -23,6 +20,16 @@ from pip._internal.network.xmlrpc import PipXmlrpcTransport
 from pip._internal.utils.compat import get_terminal_size
 from pip._internal.utils.logging import indent_log
 from pip._internal.utils.misc import write_output
+from pip._internal.utils.typing import MYPY_CHECK_RUNNING
+
+if MYPY_CHECK_RUNNING:
+    from optparse import Values
+    from typing import Any, List, Dict, Optional
+    from typing_extensions import TypedDict
+    TransformedHit = TypedDict(
+        'TransformedHit',
+        {'name': str, 'summary': str, 'versions': List[str]},
+    )
 
 logger = logging.getLogger(__name__)
 
@@ -35,6 +42,7 @@ class SearchCommand(Command, SessionCommandMixin):
     ignore_require_venv = True
 
     def __init__(self, *args, **kw):
+        # type: (*Any, **Any) -> None
         super(SearchCommand, self).__init__(*args, **kw)
         self.cmd_opts.add_option(
             '-i', '--index',
@@ -46,6 +54,7 @@ class SearchCommand(Command, SessionCommandMixin):
         self.parser.insert_option_group(0, self.cmd_opts)
 
     def run(self, options, args):
+        # type: (Values, List[str]) -> int
         if not args:
             raise CommandError('Missing required argument (search query).')
         query = args
@@ -62,6 +71,7 @@ class SearchCommand(Command, SessionCommandMixin):
         return NO_MATCHES_FOUND
 
     def search(self, query, options):
+        # type: (List[str], Values) -> List[Dict[str, str]]
         index_url = options.index
 
         session = self.get_default_session(options)
@@ -73,12 +83,13 @@ class SearchCommand(Command, SessionCommandMixin):
 
 
 def transform_hits(hits):
+    # type: (List[Dict[str, str]]) -> List[TransformedHit]
     """
     The list from pypi is really a list of versions. We want a list of
     packages with the list of versions stored inline. This converts the
     list from pypi into one we can use.
     """
-    packages = OrderedDict()
+    packages = OrderedDict()  # type: OrderedDict[str, TransformedHit]
     for hit in hits:
         name = hit['name']
         summary = hit['summary']
@@ -101,6 +112,7 @@ def transform_hits(hits):
 
 
 def print_results(hits, name_column_width=None, terminal_width=None):
+    # type: (List[TransformedHit], Optional[int], Optional[int]) -> None
     if not hits:
         return
     if name_column_width is None:
@@ -118,8 +130,9 @@ def print_results(hits, name_column_width=None, terminal_width=None):
             target_width = terminal_width - name_column_width - 5
             if target_width > 10:
                 # wrap and indent summary to fit terminal
-                summary = textwrap.wrap(summary, target_width)
+                summary_lines = textwrap.wrap(summary, target_width)
-                summary = ('\n' + ' ' * (name_column_width + 3)).join(summary)
+                summary = ('\n' + ' ' * (name_column_width + 3)).join(
+                    summary_lines)
 
         line = '{name_latest:{name_column_width}} - {summary}'.format(
             name_latest='{name} ({latest})'.format(**locals()),
@@ -143,4 +156,5 @@ def print_results(hits, name_column_width=None, terminal_width=None):
 
 
 def highest_version(versions):
+    # type: (List[str]) -> str
     return max(versions, key=parse_version)