Do not use PyPI.org JSON API by default

* Add new CLI flag "--use-pypi-json-api" to use exclusively the PyPI
  fast but incorrect JSON API.
* Remove pypi_simple_repo_in_repos() from 'python_inspector.resolution'
  This is not longer needed
* Also remove dangerous globals from function and methods arguments:
  * PythonInputProvider.__init__() where repos was a mutable list
  * get_resolved_dependencies() where repos was a mutable list
* Refactor methods that dealt with PyPI JSON vs. simple by splitting
  them in two specialized methods

Signed-off-by: Philippe Ombredanne <pombredanne@nexb.com>

Author: pombredanne

src/python_inspector/resolution.py

@@ -11,6 +11,7 @@
 import operator
 import os
 from typing import List
+from typing import Sequence
 
 import packaging.requirements
 import packaging.utils
@@ -51,9 +52,9 @@ def is_valid_version(parsed_version, requirements, identifier, bad_versions):
 
 
 class PythonInputProvider(AbstractProvider):
-    def __init__(self, environment=None, repos=[]):
+    def __init__(self, environment=None, repos=tuple()):
         self.environment = environment
-        self.repos = repos
+        self.repos = repos or []
         self.versions_by_package = {}
         self.dependencies_by_purl = {}
 
@@ -82,54 +83,75 @@ def get_versions_for_package(self, name, repo=None):
         """
         Return a list of versions for a package.
         """
-        versions = []
         if repo and self.environment:
-            for version, package in repo._get_package_versions_map(name).items():
-                wheels = package.get_supported_wheels(environment=self.environment)
-                if list(wheels):
-                    versions.append(version)
+            return self.get_versions_for_package_from_repo(name, repo)
         else:
-            if name not in self.versions_by_package:
-                api_url = f"https://pypi.org/pypi/{name}/json"
-                resp = get_response(api_url)
-                if not resp:
-                    self.versions_by_package[name] = []
-                releases = resp.get("releases") or {}
-                self.versions_by_package[name] = releases.keys() or []
-            versions = self.versions_by_package[name]
+            return self.get_versions_for_package_from_pypi_json_api(name)
+
+    def get_versions_for_package_from_repo(self, name, repo):
+        """
+        Return a list of versions for a package name from a repo
+        """
+        versions = []
+        for version, package in repo._get_package_versions_map(name).items():
+            wheels = package.get_supported_wheels(environment=self.environment)
+            if list(wheels):
+                versions.append(version)
+        return versions
+
+    def get_versions_for_package_from_pypi_json_api(self, name):
+        """
+        Return a list of versions for a package name from the PyPI.org JSON API
+        """
+        if name not in self.versions_by_package:
+            api_url = f"https://pypi.org/pypi/{name}/json"
+            resp = get_response(api_url)
+            if not resp:
+                self.versions_by_package[name] = []
+            releases = resp.get("releases") or {}
+            self.versions_by_package[name] = releases.keys() or []
+        versions = self.versions_by_package[name]
         return versions
 
     def get_requirements_for_package(self, purl, candidate):
         """
         Yield requirements for a package.
         """
         if self.repos and self.environment:
-            wheels = utils_pypi.download_wheel(
-                name=candidate.name,
-                version=str(candidate.version),
-                environment=self.environment,
-                repos=self.repos,
-            )
-            for wheel in wheels:
-                deps = list(
-                    PypiWheelHandler.parse(os.path.join(utils_pypi.CACHE_THIRDPARTY_DIR, wheel))
-                )
-                assert len(deps) == 1
-                deps = deps[0].dependencies
-                for dep in deps:
-                    if dep.scope == "install":
-                        yield packaging.requirements.Requirement(str(dep.extracted_requirement))
+            return self.get_requirements_for_package_from_pypi_simple(candidate)
         else:
-            if str(purl) not in self.dependencies_by_purl:
-                api_url = f"https://pypi.org/pypi/{purl.name}/{purl.version}/json"
-                resp = get_response(api_url)
-                if not resp:
-                    self.dependencies_by_purl[str(purl)] = []
-                info = resp.get("info") or {}
-                requires_dist = info.get("requires_dist") or []
-                self.dependencies_by_purl[str(purl)] = requires_dist
-            for dependency in self.dependencies_by_purl[str(purl)]:
-                yield packaging.requirements.Requirement(dependency)
+            return self.get_requirements_for_package_from_pypi_json_api(purl)
+
+    def get_requirements_for_package_from_pypi_simple(self, candidate):
+        wheels = utils_pypi.download_wheel(
+            name=candidate.name,
+            version=str(candidate.version),
+            environment=self.environment,
+            repos=self.repos,
+        )
+        for wheel in wheels:
+            deps = list(
+                PypiWheelHandler.parse(os.path.join(utils_pypi.CACHE_THIRDPARTY_DIR, wheel))
+            )
+            assert len(deps) == 1
+            deps = deps[0].dependencies
+            for dep in deps:
+                if dep.scope == "install":
+                    yield packaging.requirements.Requirement(str(dep.extracted_requirement))
+
+    def get_requirements_for_package_from_pypi_json_api(self, purl):
+
+        # if no repos are provided use the incorrect but fast JSON API
+        if str(purl) not in self.dependencies_by_purl:
+            api_url = f"https://pypi.org/pypi/{purl.name}/{purl.version}/json"
+            resp = get_response(api_url)
+            if not resp:
+                self.dependencies_by_purl[str(purl)] = []
+            info = resp.get("info") or {}
+            requires_dist = info.get("requires_dist") or []
+            self.dependencies_by_purl[str(purl)] = requires_dist
+        for dependency in self.dependencies_by_purl[str(purl)]:
+            yield packaging.requirements.Requirement(dependency)
 
     def get_candidates(self, all_versions, requirements, identifier, bad_versions, name, extras):
         """
@@ -267,29 +289,22 @@ def format_resolution(results, as_tree=False):
         return dependencies
 
 
-def pypi_simple_repo_in_repos(repos: utils_pypi.PypiSimpleRepository):
-    """
-    Return True if simple pypi index_url is present in any of the repos
-    """
-    return any(repo.index_url == utils_pypi.PYPI_SIMPLE_URL for repo in repos)
-
-
 def get_resolved_dependencies(
     requirements: List[Requirement],
     environment: utils_pypi.Environment = None,
-    repos: List[utils_pypi.PypiSimpleRepository] = [],
+    repos: Sequence[utils_pypi.PypiSimpleRepository] = tuple(),
     as_tree: bool = False,
 ):
     """
     Return resolved dependencies of a ``requirements`` list of Requirement for
     an ``enviroment`` Environment. The resolved dependencies are formatted as
-    parent/children or a nested tree if ``as_tree`` is True
-    """
-    if repos and not pypi_simple_repo_in_repos(repos):
-        repos.append(utils_pypi.PYPI_PUBLIC_REPO)
+    parent/children or a nested tree if ``as_tree`` is True.
 
+    Used the provided ``repos`` list of PypiSimpleRepository.
+    If empty, use instead the PyPI.org JSON API exclusively instead
+    """
     resolver = Resolver(
-        provider=PythonInputProvider(environment, repos),
+        provider=PythonInputProvider(environment=environment, repos=repos),
         reporter=BaseReporter(),
     )
     results = resolver.resolve(requirements=requirements)

src/python_inspector/resolve_cli.py

@@ -94,7 +94,13 @@
     "--use-cached-index",
     is_flag=True,
     hidden=True,
-    help="Use cached on-disk PyPI package indexes and do not refetch if present.",
+    help="Use cached on-disk PyPI simple package indexes and do not refetch if present.",
+)
+@click.option(
+    "--use-pypi-json-api",
+    is_flag=True,
+    help="Use PyPI JSON API to fetch dependency data. Faster but not always correct. "
+    "--index-url are ignored when this option is active.",
 )
 @click.option(
     "--debug",
@@ -111,6 +117,7 @@ def resolve_dependencies(
     index_urls,
     json_output,
     use_cached_index=False,
+    use_pypi_json_api=False,
     debug=TRACE,
 ):
     """
@@ -126,16 +133,10 @@ def resolve_dependencies(
 
     Error and progress are printed to stderr.
 
-    1) Without an --index-url is provided, this tool uses the PyPI JSON API.
-
     For example, display the results of resolving the dependencies for flask==2.1.2
     on screen::
 
         dad --spec "flask==2.1.2" --json -
-
-    2) If an --index-url is provided, it is used to resolve the dependencies::
-
-         dad --spec "flask==2.1.2" --index-url https://pypi.org/simple --json -
     """
 
     click.secho(f"Resolving dependencies...")
@@ -158,35 +159,36 @@ def resolve_dependencies(
     if debug:
         click.secho("direct_dependencies:")
         for dep in direct_dependencies:
-            click.secho(" ", dep)
+            click.secho(f" {dep}")
 
     # create a resolution environments
     environment = utils_pypi.Environment.from_pyver_and_os(
         python_version=python_version, operating_system=operating_system
     )
 
     if debug:
-        click.secho("environment:", environment)
+        click.secho(f"environment: {environment}")
 
-    # Collect PyPI repos
     repos = []
-    for index_url in index_urls:
-        index_url = index_url.strip("/")
-        existing = utils_pypi.DEFAULT_PYPI_REPOS_BY_URL.get(index_url)
-        if existing:
-            existing.use_cached_index = use_cached_index
-            repos.append(existing)
-        else:
-            repo = utils_pypi.PypiSimpleRepository(
-                index_url=index_url,
-                use_cached_index=use_cached_index,
-            )
-            repos.append(repo)
+    if not use_pypi_json_api:
+        # Collect PyPI repos
+        for index_url in index_urls:
+            index_url = index_url.strip("/")
+            existing = utils_pypi.DEFAULT_PYPI_REPOS_BY_URL.get(index_url)
+            if existing:
+                existing.use_cached_index = use_cached_index
+                repos.append(existing)
+            else:
+                repo = utils_pypi.PypiSimpleRepository(
+                    index_url=index_url,
+                    use_cached_index=use_cached_index,
+                )
+                repos.append(repo)
 
     if debug:
         click.secho("repos:")
         for repo in repos:
-            click.secho(" ", repo)
+            click.secho(f" {repo}")
 
     # resolve dependencies proper
     requirements, resolved_dependencies = resolve(
@@ -230,11 +232,13 @@ def resolve_dependencies(
         click.secho("done!")
 
 
-def resolve(direct_dependencies, environment, repos, as_tree=False):
+def resolve(direct_dependencies, environment, repos=tuple(), as_tree=False):
     """
     Resolve dependencies given a ``direct_dependencies`` list of
     DependentPackage and return a tuple of (initial_requirements,
     resolved_dependencies).
+    Used the provided ``repos`` list of PypiSimpleRepository.
+    If empty, use instead the PyPI.org JSON API exclusively.
     """
 
     requirements = [

tests/test_cli.py

@@ -29,9 +29,13 @@
 def test_cli_with_default_urls():
     expected_file = test_env.get_test_loc("default-url-expected.json", must_exist=False)
     specifier = "zipp==3.8.0"
+    extra_options = [
+        "--use-pypi-json-api",
+    ]
     check_specs_resolution(
         specifier=specifier,
         expected_file=expected_file,
+        extra_options=extra_options,
         regen=REGEN_TEST_FIXTURES,
     )
 

tests/test_resolution.py

@@ -14,7 +14,6 @@
 
 from python_inspector.resolution import get_resolved_dependencies
 from python_inspector.resolution import is_valid_version
-from python_inspector.resolution import pypi_simple_repo_in_repos
 from python_inspector.utils_pypi import PYPI_PUBLIC_REPO
 from python_inspector.utils_pypi import Environment
 
@@ -87,14 +86,6 @@ def test_get_resolved_dependencies_with_tilde_requirement_using_json_api():
     ]
 
 
-def test_pypi_simple_repo_in_repos():
-    assert pypi_simple_repo_in_repos(repos=[PYPI_PUBLIC_REPO])
-
-
-def test_pypi_simple_repo_in_repos_not_present():
-    assert not pypi_simple_repo_in_repos(repos=[])
-
-
 def test_is_valid_version():
     parsed_version = packaging.version.parse("2.1.2")
     requirements = {"flask": [Requirement("flask>2.0.0")]}