Update pip to 23.2.1 pt release.

pipenv/patched/patched.txt

@@ -1,2 +1,2 @@
-pip==23.2
+pip==23.2.1
 safety==2.3.2

pipenv/patched/pip/__init__.py

@@ -1,6 +1,6 @@
 from typing import List, Optional
 
-__version__ = "23.2"
+__version__ = "23.2.1"
 
 
 def main(args: Optional[List[str]] = None) -> int:

pipenv/patched/pip/_internal/cli/req_command.py

@@ -287,6 +287,7 @@ class RequirementCommand(IndexGroupCommand):
         """
         temp_build_dir_path = temp_build_dir.path
         assert temp_build_dir_path is not None
+        legacy_resolver = False
 
         resolver_variant = cls.determine_resolver_variant(options)
         if resolver_variant == "2020-resolver":
@@ -300,6 +301,7 @@ class RequirementCommand(IndexGroupCommand):
                     "production."
                 )
         else:
+            legacy_resolver = True
             lazy_wheel = False
             if "fast-deps" in options.features_enabled:
                 logger.warning(
@@ -320,6 +322,7 @@ class RequirementCommand(IndexGroupCommand):
             use_user_site=use_user_site,
             lazy_wheel=lazy_wheel,
             verbosity=verbosity,
+            legacy_resolver=legacy_resolver,
         )
 
     @classmethod

pipenv/patched/pip/_internal/network/session.py

@@ -419,15 +419,17 @@ class PipSession(requests.Session):
                 msg += f" (from {source})"
             logger.info(msg)
 
-        host_port = parse_netloc(host)
-        if host_port not in self.pip_trusted_origins:
-            self.pip_trusted_origins.append(host_port)
+        parsed_host, parsed_port = parse_netloc(host)
+        if parsed_host is None:
+            raise ValueError(f"Trusted host URL must include a host part: {host!r}")
+        if (parsed_host, parsed_port) not in self.pip_trusted_origins:
+            self.pip_trusted_origins.append((parsed_host, parsed_port))
 
         self.mount(
             build_url_from_netloc(host, scheme="http") + "/", self._trusted_host_adapter
         )
         self.mount(build_url_from_netloc(host) + "/", self._trusted_host_adapter)
-        if not host_port[1]:
+        if not parsed_port:
             self.mount(
                 build_url_from_netloc(host, scheme="http") + ":",
                 self._trusted_host_adapter,

pipenv/patched/pip/_internal/operations/prepare.py

@@ -226,6 +226,7 @@ class RequirementPreparer:
         use_user_site: bool,
         lazy_wheel: bool,
         verbosity: int,
+        legacy_resolver: bool,
     ) -> None:
         super().__init__()
 
@@ -259,6 +260,9 @@ class RequirementPreparer:
         # How verbose should underlying tooling be?
         self.verbosity = verbosity
 
+        # Are we using the legacy resolver?
+        self.legacy_resolver = legacy_resolver
+
         # Memoized downloaded files, as mapping of url: path.
         self._downloaded: Dict[str, str] = {}
 
@@ -365,6 +369,11 @@ class RequirementPreparer:
         self,
         req: InstallRequirement,
     ) -> Optional[BaseDistribution]:
+        if self.legacy_resolver:
+            logger.debug(
+                "Metadata-only fetching is not used in the legacy resolver",
+            )
+            return None
         if self.require_hashes:
             logger.debug(
                 "Metadata-only fetching is not used as hash checking is required",

pipenv/patched/pip/_internal/utils/misc.py

@@ -1,6 +1,3 @@
-# The following comment should be removed at some point in the future.
-# mypy: strict-optional=False
-
 import contextlib
 import errno
 import getpass
@@ -344,17 +341,18 @@ def write_output(msg: Any, *args: Any) -> None:
 
 
 class StreamWrapper(StringIO):
-    orig_stream: TextIO = None
+    orig_stream: TextIO
 
     @classmethod
     def from_stream(cls, orig_stream: TextIO) -> "StreamWrapper":
-        cls.orig_stream = orig_stream
-        return cls()
+        ret = cls()
+        ret.orig_stream = orig_stream
+        return ret
 
     # compileall.compile_dir() needs stdout.encoding to print to stdout
-    # https://github.com/python/mypy/issues/4125
+    # type ignore is because TextIOBase.encoding is writeable
     @property
-    def encoding(self):  # type: ignore
+    def encoding(self) -> str:  # type: ignore
         return self.orig_stream.encoding
 
 
@@ -422,7 +420,7 @@ def build_url_from_netloc(netloc: str, scheme: str = "https") -> str:
     return f"{scheme}://{netloc}"
 
 
-def parse_netloc(netloc: str) -> Tuple[str, Optional[int]]:
+def parse_netloc(netloc: str) -> Tuple[Optional[str], Optional[int]]:
     """
     Return the host-port pair from a netloc.
     """
@@ -510,7 +508,9 @@ def _redact_netloc(netloc: str) -> Tuple[str]:
     return (redact_netloc(netloc),)
 
 
-def split_auth_netloc_from_url(url: str) -> Tuple[str, str, Tuple[str, str]]:
+def split_auth_netloc_from_url(
+    url: str,
+) -> Tuple[str, str, Tuple[Optional[str], Optional[str]]]:
     """
     Parse a url into separate netloc, auth, and url with no auth.
 

pipenv/patched/pip/_vendor/urllib3/packages/backports/weakref_finalize.py

@@ -0,0 +1,155 @@
+# -*- coding: utf-8 -*-
+"""
+backports.weakref_finalize
+~~~~~~~~~~~~~~~~~~
+
+Backports the Python 3 ``weakref.finalize`` method.
+"""
+from __future__ import absolute_import
+
+import itertools
+import sys
+from weakref import ref
+
+__all__ = ["weakref_finalize"]
+
+
+class weakref_finalize(object):
+    """Class for finalization of weakrefable objects
+    finalize(obj, func, *args, **kwargs) returns a callable finalizer
+    object which will be called when obj is garbage collected. The
+    first time the finalizer is called it evaluates func(*arg, **kwargs)
+    and returns the result. After this the finalizer is dead, and
+    calling it just returns None.
+    When the program exits any remaining finalizers for which the
+    atexit attribute is true will be run in reverse order of creation.
+    By default atexit is true.
+    """
+
+    # Finalizer objects don't have any state of their own.  They are
+    # just used as keys to lookup _Info objects in the registry.  This
+    # ensures that they cannot be part of a ref-cycle.
+
+    __slots__ = ()
+    _registry = {}
+    _shutdown = False
+    _index_iter = itertools.count()
+    _dirty = False
+    _registered_with_atexit = False
+
+    class _Info(object):
+        __slots__ = ("weakref", "func", "args", "kwargs", "atexit", "index")
+
+    def __init__(self, obj, func, *args, **kwargs):
+        if not self._registered_with_atexit:
+            # We may register the exit function more than once because
+            # of a thread race, but that is harmless
+            import atexit
+
+            atexit.register(self._exitfunc)
+            weakref_finalize._registered_with_atexit = True
+        info = self._Info()
+        info.weakref = ref(obj, self)
+        info.func = func
+        info.args = args
+        info.kwargs = kwargs or None
+        info.atexit = True
+        info.index = next(self._index_iter)
+        self._registry[self] = info
+        weakref_finalize._dirty = True
+
+    def __call__(self, _=None):
+        """If alive then mark as dead and return func(*args, **kwargs);
+        otherwise return None"""
+        info = self._registry.pop(self, None)
+        if info and not self._shutdown:
+            return info.func(*info.args, **(info.kwargs or {}))
+
+    def detach(self):
+        """If alive then mark as dead and return (obj, func, args, kwargs);
+        otherwise return None"""
+        info = self._registry.get(self)
+        obj = info and info.weakref()
+        if obj is not None and self._registry.pop(self, None):
+            return (obj, info.func, info.args, info.kwargs or {})
+
+    def peek(self):
+        """If alive then return (obj, func, args, kwargs);
+        otherwise return None"""
+        info = self._registry.get(self)
+        obj = info and info.weakref()
+        if obj is not None:
+            return (obj, info.func, info.args, info.kwargs or {})
+
+    @property
+    def alive(self):
+        """Whether finalizer is alive"""
+        return self in self._registry
+
+    @property
+    def atexit(self):
+        """Whether finalizer should be called at exit"""
+        info = self._registry.get(self)
+        return bool(info) and info.atexit
+
+    @atexit.setter
+    def atexit(self, value):
+        info = self._registry.get(self)
+        if info:
+            info.atexit = bool(value)
+
+    def __repr__(self):
+        info = self._registry.get(self)
+        obj = info and info.weakref()
+        if obj is None:
+            return "<%s object at %#x; dead>" % (type(self).__name__, id(self))
+        else:
+            return "<%s object at %#x; for %r at %#x>" % (
+                type(self).__name__,
+                id(self),
+                type(obj).__name__,
+                id(obj),
+            )
+
+    @classmethod
+    def _select_for_exit(cls):
+        # Return live finalizers marked for exit, oldest first
+        L = [(f, i) for (f, i) in cls._registry.items() if i.atexit]
+        L.sort(key=lambda item: item[1].index)
+        return [f for (f, i) in L]
+
+    @classmethod
+    def _exitfunc(cls):
+        # At shutdown invoke finalizers for which atexit is true.
+        # This is called once all other non-daemonic threads have been
+        # joined.
+        reenable_gc = False
+        try:
+            if cls._registry:
+                import gc
+
+                if gc.isenabled():
+                    reenable_gc = True
+                    gc.disable()
+                pending = None
+                while True:
+                    if pending is None or weakref_finalize._dirty:
+                        pending = cls._select_for_exit()
+                        weakref_finalize._dirty = False
+                    if not pending:
+                        break
+                    f = pending.pop()
+                    try:
+                        # gc is disabled, so (assuming no daemonic
+                        # threads) the following is the only line in
+                        # this function which might trigger creation
+                        # of a new finalizer
+                        f()
+                    except Exception:
+                        sys.excepthook(*sys.exc_info())
+                    assert f not in cls._registry
+        finally:
+            # prevent any more finalizers from executing during shutdown
+            weakref_finalize._shutdown = True
+            if reenable_gc:
+                gc.enable()