Merge pull request #5939 from cytolentino/internal-cache-typing

Add mypy annotations to pip._internal.cache

src/pip/_internal/cache.py

@@ -12,8 +12,13 @@
 from pip._internal.models.link import Link
 from pip._internal.utils.compat import expanduser
 from pip._internal.utils.temp_dir import TempDirectory
+from pip._internal.utils.typing import MYPY_CHECK_RUNNING
 from pip._internal.wheel import InvalidWheelFilename, Wheel
 
+if MYPY_CHECK_RUNNING:
+    from typing import Optional, Set, List, Any  # noqa: F401
+    from pip._internal.index import FormatControl  # noqa: F401
+
 logger = logging.getLogger(__name__)
 
 
@@ -29,6 +34,7 @@ class Cache(object):
     """
 
     def __init__(self, cache_dir, format_control, allowed_formats):
+        # type: (str, FormatControl, Set[str]) -> None
         super(Cache, self).__init__()
         self.cache_dir = expanduser(cache_dir) if cache_dir else None
         self.format_control = format_control
@@ -38,6 +44,7 @@ def __init__(self, cache_dir, format_control, allowed_formats):
         assert self.allowed_formats.union(_valid_formats) == _valid_formats
 
     def _get_cache_path_parts(self, link):
+        # type: (Link) -> List[str]
         """Get parts of part that must be os.path.joined with cache_dir
         """
 
@@ -63,6 +70,7 @@ def _get_cache_path_parts(self, link):
         return parts
 
     def _get_candidates(self, link, package_name):
+        # type: (Link, Optional[str]) -> List[Any]
         can_not_cache = (
             not self.cache_dir or
             not package_name or
@@ -87,23 +95,27 @@ def _get_candidates(self, link, package_name):
             raise
 
     def get_path_for_link(self, link):
+        # type: (Link) -> str
         """Return a directory to store cached items in for link.
         """
         raise NotImplementedError()
 
     def get(self, link, package_name):
+        # type: (Link, Optional[str]) -> Link
         """Returns a link to a cached item if it exists, otherwise returns the
         passed link.
         """
         raise NotImplementedError()
 
     def _link_for_candidate(self, link, candidate):
+        # type: (Link, str) -> Link
         root = self.get_path_for_link(link)
         path = os.path.join(root, candidate)
 
         return Link(path_to_url(path))
 
     def cleanup(self):
+        # type: () -> None
         pass
 
 
@@ -112,11 +124,13 @@ class SimpleWheelCache(Cache):
     """
 
     def __init__(self, cache_dir, format_control):
+        # type: (str, FormatControl) -> None
         super(SimpleWheelCache, self).__init__(
             cache_dir, format_control, {"binary"}
         )
 
     def get_path_for_link(self, link):
+        # type: (Link) -> str
         """Return a directory to store cached wheels for link
 
         Because there are M wheels for any one sdist, we provide a directory
@@ -137,6 +151,7 @@ def get_path_for_link(self, link):
         return os.path.join(self.cache_dir, "wheels", *parts)
 
     def get(self, link, package_name):
+        # type: (Link, Optional[str]) -> Link
         candidates = []
 
         for wheel_name in self._get_candidates(link, package_name):
@@ -160,6 +175,7 @@ class EphemWheelCache(SimpleWheelCache):
     """
 
     def __init__(self, format_control):
+        # type: (FormatControl) -> None
         self._temp_dir = TempDirectory(kind="ephem-wheel-cache")
         self._temp_dir.create()
 
@@ -168,6 +184,7 @@ def __init__(self, format_control):
         )
 
     def cleanup(self):
+        # type: () -> None
         self._temp_dir.cleanup()
 
 
@@ -179,24 +196,29 @@ class WheelCache(Cache):
     """
 
     def __init__(self, cache_dir, format_control):
+        # type: (str, FormatControl) -> None
         super(WheelCache, self).__init__(
             cache_dir, format_control, {'binary'}
         )
         self._wheel_cache = SimpleWheelCache(cache_dir, format_control)
         self._ephem_cache = EphemWheelCache(format_control)
 
     def get_path_for_link(self, link):
+        # type: (Link) -> str
         return self._wheel_cache.get_path_for_link(link)
 
     def get_ephem_path_for_link(self, link):
+        # type: (Link) -> str
         return self._ephem_cache.get_path_for_link(link)
 
     def get(self, link, package_name):
+        # type: (Link, Optional[str]) -> Link
         retval = self._wheel_cache.get(link, package_name)
         if retval is link:
             retval = self._ephem_cache.get(link, package_name)
         return retval
 
     def cleanup(self):
+        # type: () -> None
         self._wheel_cache.cleanup()
         self._ephem_cache.cleanup()