Complete type annotations in pip/_internal/network (#10184)

Author: harupy

src/pip/_internal/network/auth.py

@@ -36,8 +36,7 @@
     keyring = None
 
 
-def get_keyring_auth(url, username):
-    # type: (Optional[str], Optional[str]) -> Optional[AuthInfo]
+def get_keyring_auth(url: Optional[str], username: Optional[str]) -> Optional[AuthInfo]:
     """Return the tuple auth for a given url from keyring."""
     global keyring
     if not url or not keyring:
@@ -71,20 +70,20 @@ def get_keyring_auth(url, username):
 
 class MultiDomainBasicAuth(AuthBase):
 
-    def __init__(self, prompting=True, index_urls=None):
-        # type: (bool, Optional[List[str]]) -> None
+    def __init__(
+        self, prompting: bool = True, index_urls: Optional[List[str]] = None
+    ) -> None:
         self.prompting = prompting
         self.index_urls = index_urls
-        self.passwords = {}  # type: Dict[str, AuthInfo]
+        self.passwords: Dict[str, AuthInfo] = {}
         # When the user is prompted to enter credentials and keyring is
         # available, we will offer to save them. If the user accepts,
         # this value is set to the credentials they entered. After the
         # request authenticates, the caller should call
         # ``save_credentials`` to save these.
-        self._credentials_to_save = None  # type: Optional[Credentials]
+        self._credentials_to_save: Optional[Credentials] = None
 
-    def _get_index_url(self, url):
-        # type: (str) -> Optional[str]
+    def _get_index_url(self, url: str) -> Optional[str]:
         """Return the original index URL matching the requested URL.
 
         Cached or dynamically generated credentials may work against
@@ -106,9 +105,8 @@ def _get_index_url(self, url):
                 return u
         return None
 
-    def _get_new_credentials(self, original_url, allow_netrc=True,
-                             allow_keyring=False):
-        # type: (str, bool, bool) -> AuthInfo
+    def _get_new_credentials(self, original_url: str, allow_netrc: bool = True,
+                             allow_keyring: bool = False) -> AuthInfo:
         """Find and return credentials for the specified URL."""
         # Split the credentials and netloc from the url.
         url, netloc, url_user_password = split_auth_netloc_from_url(
@@ -157,8 +155,9 @@ def _get_new_credentials(self, original_url, allow_netrc=True,
 
         return username, password
 
-    def _get_url_and_credentials(self, original_url):
-        # type: (str) -> Tuple[str, Optional[str], Optional[str]]
+    def _get_url_and_credentials(
+        self, original_url: str
+    ) -> Tuple[str, Optional[str], Optional[str]]:
         """Return the credentials to use for the provided URL.
 
         If allowed, netrc and keyring may be used to obtain the
@@ -197,8 +196,7 @@ def _get_url_and_credentials(self, original_url):
 
         return url, username, password
 
-    def __call__(self, req):
-        # type: (Request) -> Request
+    def __call__(self, req: Request) -> Request:
         # Get credentials for this request
         url, username, password = self._get_url_and_credentials(req.url)
 
@@ -215,8 +213,9 @@ def __call__(self, req):
         return req
 
     # Factored out to allow for easy patching in tests
-    def _prompt_for_password(self, netloc):
-        # type: (str) -> Tuple[Optional[str], Optional[str], bool]
+    def _prompt_for_password(
+        self, netloc: str
+    ) -> Tuple[Optional[str], Optional[str], bool]:
         username = ask_input(f"User for {netloc}: ")
         if not username:
             return None, None, False
@@ -227,14 +226,12 @@ def _prompt_for_password(self, netloc):
         return username, password, True
 
     # Factored out to allow for easy patching in tests
-    def _should_save_password_to_keyring(self):
-        # type: () -> bool
+    def _should_save_password_to_keyring(self) -> bool:
         if not keyring:
             return False
         return ask("Save credentials to keyring [y/N]: ", ["y", "n"]) == "y"
 
-    def handle_401(self, resp, **kwargs):
-        # type: (Response, **Any) -> Response
+    def handle_401(self, resp: Response, **kwargs: Any) -> Response:
         # We only care about 401 responses, anything else we want to just
         #   pass through the actual response
         if resp.status_code != 401:
@@ -286,16 +283,14 @@ def handle_401(self, resp, **kwargs):
 
         return new_resp
 
-    def warn_on_401(self, resp, **kwargs):
-        # type: (Response, **Any) -> None
+    def warn_on_401(self, resp: Response, **kwargs: Any) -> None:
         """Response callback to warn about incorrect credentials."""
         if resp.status_code == 401:
             logger.warning(
                 '401 Error, Credentials not correct for %s', resp.request.url,
             )
 
-    def save_credentials(self, resp, **kwargs):
-        # type: (Response, **Any) -> None
+    def save_credentials(self, resp: Response, **kwargs: Any) -> None:
         """Response callback to save credentials on success."""
         assert keyring is not None, "should never reach here without keyring"
         if not keyring:

src/pip/_internal/network/cache.py

@@ -13,14 +13,12 @@
 from pip._internal.utils.misc import ensure_dir
 
 
-def is_from_cache(response):
-    # type: (Response) -> bool
+def is_from_cache(response: Response) -> bool:
     return getattr(response, "from_cache", False)
 
 
 @contextmanager
-def suppressed_cache_errors():
-    # type: () -> Iterator[None]
+def suppressed_cache_errors() -> Iterator[None]:
     """If we can't access the cache then we can just skip caching and process
     requests as if caching wasn't enabled.
     """
@@ -36,30 +34,26 @@ class SafeFileCache(BaseCache):
     not be accessible or writable.
     """
 
-    def __init__(self, directory):
-        # type: (str) -> None
+    def __init__(self, directory: str) -> None:
         assert directory is not None, "Cache directory must not be None."
         super().__init__()
         self.directory = directory
 
-    def _get_cache_path(self, name):
-        # type: (str) -> str
+    def _get_cache_path(self, name: str) -> str:
         # From cachecontrol.caches.file_cache.FileCache._fn, brought into our
         # class for backwards-compatibility and to avoid using a non-public
         # method.
         hashed = FileCache.encode(name)
         parts = list(hashed[:5]) + [hashed]
         return os.path.join(self.directory, *parts)
 
-    def get(self, key):
-        # type: (str) -> Optional[bytes]
+    def get(self, key: str) -> Optional[bytes]:
         path = self._get_cache_path(key)
         with suppressed_cache_errors():
             with open(path, 'rb') as f:
                 return f.read()
 
-    def set(self, key, value):
-        # type: (str, bytes) -> None
+    def set(self, key: str, value: bytes) -> None:
         path = self._get_cache_path(key)
         with suppressed_cache_errors():
             ensure_dir(os.path.dirname(path))
@@ -69,8 +63,7 @@ def set(self, key, value):
 
             replace(f.name, path)
 
-    def delete(self, key):
-        # type: (str) -> None
+    def delete(self, key: str) -> None:
         path = self._get_cache_path(key)
         with suppressed_cache_errors():
             os.remove(path)

src/pip/_internal/network/download.py

@@ -20,20 +20,18 @@
 logger = logging.getLogger(__name__)
 
 
-def _get_http_response_size(resp):
-    # type: (Response) -> Optional[int]
+def _get_http_response_size(resp: Response) -> Optional[int]:
     try:
         return int(resp.headers['content-length'])
     except (ValueError, KeyError, TypeError):
         return None
 
 
 def _prepare_download(
-    resp,  # type: Response
-    link,  # type: Link
-    progress_bar  # type: str
-):
-    # type: (...) -> Iterable[bytes]
+    resp: Response,
+    link: Link,
+    progress_bar: str
+) -> Iterable[bytes]:
     total_length = _get_http_response_size(resp)
 
     if link.netloc == PyPI.file_storage_domain:
@@ -72,16 +70,14 @@ def _prepare_download(
     )(chunks)
 
 
-def sanitize_content_filename(filename):
-    # type: (str) -> str
+def sanitize_content_filename(filename: str) -> str:
     """
     Sanitize the "filename" value from a Content-Disposition header.
     """
     return os.path.basename(filename)
 
 
-def parse_content_disposition(content_disposition, default_filename):
-    # type: (str, str) -> str
+def parse_content_disposition(content_disposition: str, default_filename: str) -> str:
     """
     Parse the "filename" value from a Content-Disposition header, and
     return the default filename if the result is empty.
@@ -95,8 +91,7 @@ def parse_content_disposition(content_disposition, default_filename):
     return filename or default_filename
 
 
-def _get_http_response_filename(resp, link):
-    # type: (Response, Link) -> str
+def _get_http_response_filename(resp: Response, link: Link) -> str:
     """Get an ideal filename from the given HTTP response, falling back to
     the link filename if not provided.
     """
@@ -105,7 +100,7 @@ def _get_http_response_filename(resp, link):
     content_disposition = resp.headers.get('content-disposition')
     if content_disposition:
         filename = parse_content_disposition(content_disposition, filename)
-    ext = splitext(filename)[1]  # type: Optional[str]
+    ext: Optional[str] = splitext(filename)[1]
     if not ext:
         ext = mimetypes.guess_extension(
             resp.headers.get('content-type', '')
@@ -119,8 +114,7 @@ def _get_http_response_filename(resp, link):
     return filename
 
 
-def _http_get_download(session, link):
-    # type: (PipSession, Link) -> Response
+def _http_get_download(session: PipSession, link: Link) -> Response:
     target_url = link.url.split('#', 1)[0]
     resp = session.get(target_url, headers=HEADERS, stream=True)
     raise_for_status(resp)
@@ -130,15 +124,13 @@ def _http_get_download(session, link):
 class Downloader:
     def __init__(
         self,
-        session,  # type: PipSession
-        progress_bar,  # type: str
-    ):
-        # type: (...) -> None
+        session: PipSession,
+        progress_bar: str,
+    ) -> None:
         self._session = session
         self._progress_bar = progress_bar
 
-    def __call__(self, link, location):
-        # type: (Link, str) -> Tuple[str, str]
+    def __call__(self, link: Link, location: str) -> Tuple[str, str]:
         """Download the file given by link into location."""
         try:
             resp = _http_get_download(self._session, link)
@@ -164,15 +156,15 @@ class BatchDownloader:
 
     def __init__(
         self,
-        session,  # type: PipSession
-        progress_bar,  # type: str
-    ):
-        # type: (...) -> None
+        session: PipSession,
+        progress_bar: str,
+    ) -> None:
         self._session = session
         self._progress_bar = progress_bar
 
-    def __call__(self, links, location):
-        # type: (Iterable[Link], str) -> Iterable[Tuple[Link, Tuple[str, str]]]
+    def __call__(
+        self, links: Iterable[Link], location: str
+    ) -> Iterable[Tuple[Link, Tuple[str, str]]]:
         """Download the files given by links into location."""
         for link in links:
             try: