Replace the old Entry class with something simpler based on data-classes + pip session requests performance issue patch

Author: matteius

pipenv/resolver.py

@@ -701,7 +701,6 @@ def do_init(
         pypi_mirror=pypi_mirror,
         categories=categories,
     )
-    err.print(packages_updated)
 
     if not allow_global and not deploy and "PIPENV_ACTIVE" not in os.environ:
         console.print(

pipenv/routines/install.py

@@ -87,22 +87,18 @@ def clean_pkg_version(version):
 
 def get_lockfile_section_using_pipfile_category(category):
     if category == "dev-packages":
-        lockfile_section = "develop"
+        return "develop"
     elif category == "packages":
-        lockfile_section = "default"
-    else:
-        lockfile_section = category
-    return lockfile_section
+        return "default"
+    return category
 
 
 def get_pipfile_category_using_lockfile_section(category):
     if category == "develop":
-        lockfile_section = "dev-packages"
+        return "dev-packages"
     elif category == "default":
-        lockfile_section = "packages"
-    else:
-        lockfile_section = category
-    return lockfile_section
+        return "packages"
+    return category
 
 
 class HackedPythonVersion:
@@ -485,7 +481,7 @@ def dependency_as_pip_install_line(
         else:
             if "#egg=" in vcs_url:
                 vcs_url = vcs_url.split("#egg=")[0]
-            git_req = f"{dep_name}{extras}@ {include_vcs}{vcs_url}{ref}"
+            git_req = f"{dep_name}{extras} @ {include_vcs}{vcs_url}{ref}"
             if "subdirectory" in dep:
                 git_req += f"#subdirectory={dep['subdirectory']}"
 
@@ -780,7 +776,7 @@ def determine_package_name(package: InstallRequirement):
     elif "#egg=" in str(package):
         req_name = str(package).split("#egg=")[1]
         req_name = req_name.split("[")[0]
-    elif "@ " in str(package):
+    elif " @ " in str(package):
         req_name = str(package).split("@ ")[0]
         req_name = req_name.split("[")[0]
     elif package.link and package.link.scheme in REMOTE_SCHEMES:

pipenv/utils/dependencies.py

@@ -7,7 +7,7 @@
 from json import JSONDecodeError
 from pathlib import Path
 from tempfile import NamedTemporaryFile
-from typing import Any, Dict, Iterator, List, Optional
+from typing import Any, Dict, Iterator, List, Optional, Set, Tuple
 
 from pipenv.patched.pip._internal.req.req_install import InstallRequirement
 from pipenv.utils.dependencies import (
@@ -43,77 +43,96 @@ def merge_markers(entry, markers):
 
 def format_requirement_for_lockfile(
     req: InstallRequirement,
-    markers_lookup,
-    index_lookup,
-    original_deps,
-    pipfile_entries,
-    hashes=None,
-):
-    if req.specifier:
-        version = str(req.specifier)
-    else:
-        version = None
+    markers_lookup: Dict[str, str],
+    index_lookup: Dict[str, str],
+    original_deps: Dict[str, Any],
+    pipfile_entries: Dict[str, Any],
+    hashes: Optional[Set[str]] = None,
+) -> Tuple[str, Dict[str, Any]]:
+    """Format a requirement for the lockfile with improved VCS handling."""
     name = normalize_name(req.name)
-    index = index_lookup.get(name)
-    markers = req.markers
-    req.index = index
-    pipfile_entry = pipfile_entries[name] if name in pipfile_entries else {}
-    entry = {}
+    entry: Dict[str, Any] = {"name": name}
+    pipfile_entry = pipfile_entries.get(name, {})
+
+    # Handle VCS requirements
     if req.link and req.link.is_vcs:
         vcs = req.link.scheme.split("+", 1)[0]
-        entry["ref"] = determine_vcs_revision_hash(req, vcs, pipfile_entry.get("ref"))
 
+        # Get VCS URL from original deps or normalize the link URL
         if name in original_deps:
             entry[vcs] = original_deps[name]
         else:
             vcs_url, _ = normalize_vcs_url(req.link.url)
             entry[vcs] = vcs_url
-        if pipfile_entry.get("subdirectory"):
+
+        # Handle reference information - try multiple sources
+        ref = determine_vcs_revision_hash(req, vcs, pipfile_entry.get("ref"))
+        if ref:
+            entry["ref"] = ref
+
+        # Handle subdirectory information
+        if isinstance(pipfile_entry, dict) and pipfile_entry.get("subdirectory"):
             entry["subdirectory"] = pipfile_entry["subdirectory"]
         elif req.link.subdirectory_fragment:
             entry["subdirectory"] = req.link.subdirectory_fragment
-    if req.req:
-        entry["version"] = str(req.specifier)
-    elif version:
-        entry["version"] = version
-    elif req.link and req.link.is_file:
-        entry["file"] = req.link.url
-    if hashes:
-        entry["hashes"] = sorted(set(hashes))
-    entry["name"] = name
-    if index:
-        entry.update({"index": index})
+
+    # Handle non-VCS requirements
+    else:
+        if req.req and req.req.specifier:
+            entry["version"] = str(req.req.specifier)
+        elif req.specifier:
+            entry["version"] = str(req.specifier)
+        elif req.link and req.link.is_file:
+            entry["file"] = req.link.url
+
+    # Add index information
+    if name in index_lookup:
+        entry["index"] = index_lookup[name]
+
+    # Handle markers
+    markers = req.markers
     if markers:
-        entry.update({"markers": str(markers)})
+        entry["markers"] = str(markers)
     if name in markers_lookup:
         merge_markers(entry, markers_lookup[name])
-    if isinstance(pipfile_entry, dict) and "markers" in pipfile_entry:
-        merge_markers(entry, pipfile_entry["markers"])
-    if isinstance(pipfile_entry, dict) and "os_name" in pipfile_entry:
-        merge_markers(entry, f"os_name {pipfile_entry['os_name']}")
-    entry = translate_markers(entry)
+    if isinstance(pipfile_entry, dict):
+        if "markers" in pipfile_entry:
+            merge_markers(entry, pipfile_entry["markers"])
+        if "os_name" in pipfile_entry:
+            merge_markers(entry, f"os_name {pipfile_entry['os_name']}")
+
+    # Handle extras
     if req.extras:
         entry["extras"] = sorted(req.extras)
-    if isinstance(pipfile_entry, dict) and pipfile_entry.get("file"):
-        entry["file"] = pipfile_entry["file"]
-        if pipfile_entry.get("editable"):
-            entry["editable"] = pipfile_entry.get("editable")
-        entry.pop("version", None)
-        entry.pop("index", None)
-    elif isinstance(pipfile_entry, dict) and pipfile_entry.get("path"):
-        entry["path"] = pipfile_entry["path"]
-        if pipfile_entry.get("editable"):
-            entry["editable"] = pipfile_entry.get("editable")
-        entry.pop("version", None)
-        entry.pop("index", None)
+
+    # Handle hashes
+    if hashes:
+        entry["hashes"] = sorted(set(hashes))
+
+    # Handle file/path entries from Pipfile
+    if isinstance(pipfile_entry, dict):
+        if pipfile_entry.get("file"):
+            entry["file"] = pipfile_entry["file"]
+            if pipfile_entry.get("editable"):
+                entry["editable"] = pipfile_entry["editable"]
+            entry.pop("version", None)
+            entry.pop("index", None)
+        elif pipfile_entry.get("path"):
+            entry["path"] = pipfile_entry["path"]
+            if pipfile_entry.get("editable"):
+                entry["editable"] = pipfile_entry["editable"]
+            entry.pop("version", None)
+            entry.pop("index", None)
+
+    entry = translate_markers(entry)
     return name, entry
 
 
 def get_locked_dep(project, dep, pipfile_section, current_entry=None):
     # initialize default values
     is_top_level = False
 
-    # if the dependency has a name, find corresponding entry in pipfile
+    # # if the dependency has a name, find corresponding entry in pipfile
     if isinstance(dep, dict) and dep.get("name"):
         dep_name = pep423_name(dep["name"])
         for pipfile_key, pipfile_entry in pipfile_section.items():

pipenv/utils/locking.py

@@ -1,6 +1,5 @@
 import os
 import sys
-from functools import lru_cache
 from typing import Optional
 
 from pipenv import exceptions
@@ -97,7 +96,6 @@ def ensure_project(
     os.environ["PIP_PYTHON_PATH"] = project.python(system=system)
 
 
-@lru_cache
 def get_setuptools_version() -> Optional["STRING_TYPE"]:
     try:
         setuptools_dist = importlib_metadata.distribution("setuptools")

pipenv/utils/project.py

@@ -193,7 +193,7 @@ def requirement_from_lockfile(
                 pip_line = f"-e {include_vcs}{vcs_url}{ref_str}{egg_fragment}{extras}"
                 pip_line += f"&subdirectory={subdirectory}" if subdirectory else ""
             else:
-                pip_line = f"{package_name}{extras}@ {include_vcs}{vcs_url}{ref_str}"
+                pip_line = f"{package_name}{extras} @ {include_vcs}{vcs_url}{ref_str}"
                 pip_line += f"#subdirectory={subdirectory}" if subdirectory else ""
             return pip_line
     # Handling file-sourced packages

pipenv/utils/requirements.py

@@ -5,9 +5,9 @@
 import sys
 import tempfile
 import warnings
-from functools import lru_cache
+from functools import cached_property, lru_cache
 from pathlib import Path
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 from pipenv import environments, resolver
 from pipenv.exceptions import ResolutionFailure
@@ -27,6 +27,7 @@
 from pipenv.patched.pip._vendor.packaging.utils import canonicalize_name
 from pipenv.project import Project
 from pipenv.utils import console, err
+from pipenv.utils.dependencies import determine_vcs_revision_hash, normalize_vcs_url
 from pipenv.utils.fileutils import create_tracked_tempdir
 from pipenv.utils.requirements import normalize_name
 
@@ -38,7 +39,6 @@
     get_lockfile_section_using_pipfile_category,
     is_pinned_requirement,
     prepare_constraint_file,
-    translate_markers,
 )
 from .indexes import parse_indexes, prepare_pip_source_args
 from .internet import is_pypi_url
@@ -308,7 +308,7 @@ def pip_options(self):
         )
         return pip_options
 
-    @property
+    @cached_property
     def session(self):
         return self.pip_command._build_session(self.pip_options)
 
@@ -589,36 +589,57 @@ def resolve_hashes(self):
         return self.hashes
 
     def clean_skipped_result(
-        self, req_name: str, ireq: InstallRequirement, pipfile_entry
-    ):
-        ref = None
+        self,
+        req_name: str,
+        ireq: InstallRequirement,
+        pipfile_entry: Union[str, Dict[str, Any]],
+    ) -> Tuple[str, Dict[str, Any]]:
+        """Clean up skipped requirements with better VCS handling."""
+        # Start with pipfile entry if it's a dict, otherwise create new dict
+        entry = pipfile_entry.copy() if isinstance(pipfile_entry, dict) else {}
+        entry["name"] = req_name
+
+        # Handle VCS references
         if ireq.link and ireq.link.is_vcs:
-            ref = ireq.link.egg_fragment
+            vcs = ireq.link.scheme.split("+", 1)[0]
 
-        if isinstance(pipfile_entry, dict):
-            entry = pipfile_entry.copy()
-        else:
-            entry = {}
-        entry["name"] = req_name
+            # Try to get reference from multiple sources
+            ref = determine_vcs_revision_hash(ireq, vcs, ireq.link)
+
+            if ref:
+                entry["ref"] = ref
+            elif ireq.link.hash:
+                entry["ref"] = ireq.link.hash
+
+            # Ensure VCS URL is present
+            if vcs not in entry:
+                vcs_url, _ = normalize_vcs_url(ireq.link.url)
+                entry[vcs] = vcs_url
+
+        # Remove version if editable
         if entry.get("editable", False) and entry.get("version"):
             del entry["version"]
-        ref = ref if ref is not None else entry.get("ref")
-        if ref:
-            entry["ref"] = ref
+
+        # Add hashes
         collected_hashes = self.collect_hashes(ireq)
         if collected_hashes:
             entry["hashes"] = sorted(set(collected_hashes))
+
         return req_name, entry
 
-    def clean_results(self):
-        reqs = [(ireq,) for ireq in self.resolved_tree]
+    def clean_results(self) -> List[Dict[str, Any]]:
+        """Clean all results including both resolved and skipped packages."""
         results = {}
-        for (ireq,) in reqs:
+
+        # Handle resolved packages
+        for ireq in self.resolved_tree:
             if normalize_name(ireq.name) in self.skipped:
                 continue
+
             collected_hashes = self.hashes.get(ireq, set())
             if collected_hashes:
                 collected_hashes = sorted(collected_hashes)
+
             name, entry = format_requirement_for_lockfile(
                 ireq,
                 self.markers_lookup,
@@ -627,23 +648,25 @@ def clean_results(self):
                 self.pipfile_entries,
                 collected_hashes,
             )
-            entry = translate_markers(entry)
+
             if name in results:
                 results[name].update(entry)
             else:
                 results[name] = entry
+
+        # Handle skipped packages
         for req_name in self.skipped:
             install_req = self.install_reqs[req_name]
-            name, entry = self.clean_skipped_result(
-                req_name, install_req, self.pipfile_entries[req_name]
-            )
-            entry = translate_markers(entry)
+            pipfile_entry = self.pipfile_entries.get(req_name, {})
+
+            name, entry = self.clean_skipped_result(req_name, install_req, pipfile_entry)
+
             if name in results:
                 results[name].update(entry)
             else:
                 results[name] = entry
-        results = list(results.values())
-        return results
+
+        return list(results.values())
 
 
 def _show_warning(message, category, filename, lineno, line):

pipenv/utils/resolver.py

@@ -588,7 +588,7 @@ def test_lock_package_with_compatible_release_specifier(pipenv_instance_private_
 @pytest.mark.install
 def test_default_lock_overwrite_dev_lock(pipenv_instance_pypi):
     with pipenv_instance_pypi() as p:
-        c = p.pipenv("install 'click==6.7'")
+        c = p.pipenv("install click==6.7")
         assert c.returncode == 0
         c = p.pipenv("install -d flask")
         assert c.returncode == 0