Merge pull request #6913 from cjerdonek/link-collector-1

Move some PackageFinder methods (preparation for PR #6910)

Author: cjerdonek

src/pip/_internal/index.py

@@ -252,6 +252,65 @@ def _get_html_page(link, session=None):
     return None
 
 
+def group_locations(locations, expand_dir=False):
+    # type: (Sequence[str], bool) -> Tuple[List[str], List[str]]
+    """
+    Divide a list of locations into two groups: "files" (archives) and "urls."
+
+    :return: A pair of lists (files, urls).
+    """
+    files = []
+    urls = []
+
+    # puts the url for the given file path into the appropriate list
+    def sort_path(path):
+        url = path_to_url(path)
+        if mimetypes.guess_type(url, strict=False)[0] == 'text/html':
+            urls.append(url)
+        else:
+            files.append(url)
+
+    for url in locations:
+
+        is_local_path = os.path.exists(url)
+        is_file_url = url.startswith('file:')
+
+        if is_local_path or is_file_url:
+            if is_local_path:
+                path = url
+            else:
+                path = url_to_path(url)
+            if os.path.isdir(path):
+                if expand_dir:
+                    path = os.path.realpath(path)
+                    for item in os.listdir(path):
+                        sort_path(os.path.join(path, item))
+                elif is_file_url:
+                    urls.append(url)
+                else:
+                    logger.warning(
+                        "Path '{0}' is ignored: "
+                        "it is a directory.".format(path),
+                    )
+            elif os.path.isfile(path):
+                sort_path(path)
+            else:
+                logger.warning(
+                    "Url '%s' is ignored: it is neither a file "
+                    "nor a directory.", url,
+                )
+        elif is_url(url):
+            # Only add url with clear scheme
+            urls.append(url)
+        else:
+            logger.warning(
+                "Url '%s' is ignored. It is either a non-existing "
+                "path or lacks a specific scheme.", url,
+            )
+
+    return files, urls
+
+
 def _check_link_requires_python(
     link,  # type: Link
     version_info,  # type: Tuple[int, int, int]
@@ -899,64 +958,6 @@ def set_allow_all_prereleases(self):
         # type: () -> None
         self._candidate_prefs.allow_all_prereleases = True
 
-    @staticmethod
-    def _sort_locations(locations, expand_dir=False):
-        # type: (Sequence[str], bool) -> Tuple[List[str], List[str]]
-        """
-        Sort locations into "files" (archives) and "urls", and return
-        a pair of lists (files,urls)
-        """
-        files = []
-        urls = []
-
-        # puts the url for the given file path into the appropriate list
-        def sort_path(path):
-            url = path_to_url(path)
-            if mimetypes.guess_type(url, strict=False)[0] == 'text/html':
-                urls.append(url)
-            else:
-                files.append(url)
-
-        for url in locations:
-
-            is_local_path = os.path.exists(url)
-            is_file_url = url.startswith('file:')
-
-            if is_local_path or is_file_url:
-                if is_local_path:
-                    path = url
-                else:
-                    path = url_to_path(url)
-                if os.path.isdir(path):
-                    if expand_dir:
-                        path = os.path.realpath(path)
-                        for item in os.listdir(path):
-                            sort_path(os.path.join(path, item))
-                    elif is_file_url:
-                        urls.append(url)
-                    else:
-                        logger.warning(
-                            "Path '{0}' is ignored: "
-                            "it is a directory.".format(path),
-                        )
-                elif os.path.isfile(path):
-                    sort_path(path)
-                else:
-                    logger.warning(
-                        "Url '%s' is ignored: it is neither a file "
-                        "nor a directory.", url,
-                    )
-            elif is_url(url):
-                # Only add url with clear scheme
-                urls.append(url)
-            else:
-                logger.warning(
-                    "Url '%s' is ignored. It is either a non-existing "
-                    "path or lacks a specific scheme.", url,
-                )
-
-        return files, urls
-
     def make_link_evaluator(self, project_name):
         # type: (str) -> LinkEvaluator
         canonical_name = canonicalize_name(project_name)
@@ -971,6 +972,63 @@ def make_link_evaluator(self, project_name):
             ignore_requires_python=self._ignore_requires_python,
         )
 
+    def _sort_links(self, links):
+        # type: (Iterable[Link]) -> List[Link]
+        """
+        Returns elements of links in order, non-egg links first, egg links
+        second, while eliminating duplicates
+        """
+        eggs, no_eggs = [], []
+        seen = set()  # type: Set[Link]
+        for link in links:
+            if link not in seen:
+                seen.add(link)
+                if link.egg_fragment:
+                    eggs.append(link)
+                else:
+                    no_eggs.append(link)
+        return no_eggs + eggs
+
+    def _log_skipped_link(self, link, reason):
+        # type: (Link, Text) -> None
+        if link not in self._logged_links:
+            # Mark this as a unicode string to prevent "UnicodeEncodeError:
+            # 'ascii' codec can't encode character" in Python 2 when
+            # the reason contains non-ascii characters.
+            #   Also, put the link at the end so the reason is more visible
+            # and because the link string is usually very long.
+            logger.debug(u'Skipping link: %s: %s', reason, link)
+            self._logged_links.add(link)
+
+    def get_install_candidate(self, link_evaluator, link):
+        # type: (LinkEvaluator, Link) -> Optional[InstallationCandidate]
+        """
+        If the link is a candidate for install, convert it to an
+        InstallationCandidate and return it. Otherwise, return None.
+        """
+        is_candidate, result = link_evaluator.evaluate_link(link)
+        if not is_candidate:
+            if result:
+                self._log_skipped_link(link, reason=result)
+            return None
+
+        return InstallationCandidate(
+            project=link_evaluator.project_name,
+            link=link,
+            # Convert the Text result to str since InstallationCandidate
+            # accepts str.
+            version=str(result),
+        )
+
+    def _package_versions(self, link_evaluator, links):
+        # type: (LinkEvaluator, Iterable[Link]) -> List[InstallationCandidate]
+        result = []
+        for link in self._sort_links(links):
+            candidate = self.get_install_candidate(link_evaluator, link)
+            if candidate is not None:
+                result.append(candidate)
+        return result
+
     def find_all_candidates(self, project_name):
         # type: (str) -> List[InstallationCandidate]
         """Find all available InstallationCandidate for project_name
@@ -983,8 +1041,8 @@ def find_all_candidates(self, project_name):
         """
         search_scope = self.search_scope
         index_locations = search_scope.get_index_urls_locations(project_name)
-        index_file_loc, index_url_loc = self._sort_locations(index_locations)
-        fl_file_loc, fl_url_loc = self._sort_locations(
+        index_file_loc, index_url_loc = group_locations(index_locations)
+        fl_file_loc, fl_url_loc = group_locations(
             self.find_links, expand_dir=True,
         )
 
@@ -1177,63 +1235,6 @@ def _get_pages(self, locations, project_name):
 
             yield page
 
-    def _sort_links(self, links):
-        # type: (Iterable[Link]) -> List[Link]
-        """
-        Returns elements of links in order, non-egg links first, egg links
-        second, while eliminating duplicates
-        """
-        eggs, no_eggs = [], []
-        seen = set()  # type: Set[Link]
-        for link in links:
-            if link not in seen:
-                seen.add(link)
-                if link.egg_fragment:
-                    eggs.append(link)
-                else:
-                    no_eggs.append(link)
-        return no_eggs + eggs
-
-    def _log_skipped_link(self, link, reason):
-        # type: (Link, Text) -> None
-        if link not in self._logged_links:
-            # Mark this as a unicode string to prevent "UnicodeEncodeError:
-            # 'ascii' codec can't encode character" in Python 2 when
-            # the reason contains non-ascii characters.
-            #   Also, put the link at the end so the reason is more visible
-            # and because the link string is usually very long.
-            logger.debug(u'Skipping link: %s: %s', reason, link)
-            self._logged_links.add(link)
-
-    def get_install_candidate(self, link_evaluator, link):
-        # type: (LinkEvaluator, Link) -> Optional[InstallationCandidate]
-        """
-        If the link is a candidate for install, convert it to an
-        InstallationCandidate and return it. Otherwise, return None.
-        """
-        is_candidate, result = link_evaluator.evaluate_link(link)
-        if not is_candidate:
-            if result:
-                self._log_skipped_link(link, reason=result)
-            return None
-
-        return InstallationCandidate(
-            project=link_evaluator.project_name,
-            link=link,
-            # Convert the Text result to str since InstallationCandidate
-            # accepts str.
-            version=str(result),
-        )
-
-    def _package_versions(self, link_evaluator, links):
-        # type: (LinkEvaluator, Iterable[Link]) -> List[InstallationCandidate]
-        result = []
-        for link in self._sort_links(links):
-            candidate = self.get_install_candidate(link_evaluator, link)
-            if candidate is not None:
-                result.append(candidate)
-        return result
-
 
 def _find_name_version_sep(fragment, canonical_name):
     # type: (str, str) -> int