diff --git a/news/5893.bugfix.rst b/news/5893.bugfix.rst new file mode 100644 index 0000000000..a730fa9007 --- /dev/null +++ b/news/5893.bugfix.rst @@ -0,0 +1,17 @@ +Fix Windows py-launcher Python discovery for full version strings such as +``python_full_version = "3.11.9"`` in Pipfile and ``pipenv --python 3.11.9`` +on the CLI. + +Two bugs were addressed: + +1. ``find_python_from_py_launcher`` incorrectly required three whitespace-separated + tokens on each ``py --list-paths`` output line, which caused every non-default + Python installation (lines without the ``*`` marker) to be silently skipped. + The parser now accepts both two-token (non-default) and three-token (default) + lines, using the last token as the executable path in both cases. + +2. ``PyLauncherFinder`` stored only the ``major.minor`` version reported by + ``py --list-paths`` (e.g. ``3.11``), so searches that required a specific + patch level (``patch=9``) never matched. The finder now queries each + discovered executable for its real version string (e.g. ``3.11.9``) so that + full-version lookups succeed. diff --git a/pipenv/utils/virtualenv.py b/pipenv/utils/virtualenv.py index 9478df1117..eb952ccc34 100644 --- a/pipenv/utils/virtualenv.py +++ b/pipenv/utils/virtualenv.py @@ -485,27 +485,31 @@ def find_python_from_py_launcher(version): if c.returncode != 0: return None - # Parse the output to find the requested version + # Parse the output to find the requested version. + # The format from `py --list-paths` is one of: + # -V:3.12 * C:\...\python.exe (default version, has *) + # -V:3.11 C:\...\python.exe (non-default, no *) + # split(None, 2) gives 3 parts for default entries and only 2 parts + # for non-default entries, so we must NOT require exactly 3 parts. for line in c.stdout.splitlines(): line = line.strip() if not line: continue - # Format is: -V:3.9 * path\to\python.exe - # We need to extract the version and path parts = line.split(None, 2) - if len(parts) < 3: + if len(parts) < 2: continue - # Extract version from -V:3.9 + # Extract version from -V:X.Y v_part = parts[0] if not v_part.startswith("-V:"): continue v = v_part[3:] # Remove -V: prefix + # Path is always the last token regardless of whether * is present + path = parts[-1] if v == version: - # Found the requested version, return the path - return parts[-1] + return path except Exception: # If anything goes wrong, fall back to other methods pass diff --git a/pipenv/vendor/importlib_metadata/__init__.py b/pipenv/vendor/importlib_metadata/__init__.py index 5c45c12680..68cee05999 100644 --- a/pipenv/vendor/importlib_metadata/__init__.py +++ b/pipenv/vendor/importlib_metadata/__init__.py @@ -35,17 +35,18 @@ NullFinder, install, ) +from ._context import ExceptionTrap from ._functools import method_cache, noop, pass_none, passthrough from ._itertools import always_iterable, bucket, unique_everseen from ._meta import PackageMetadata, SimplePath -from ._typing import md_none -from .compat import py39, py311 +from .compat import py311 __all__ = [ 'Distribution', 'DistributionFinder', 'PackageMetadata', 'PackageNotFoundError', + 'MetadataNotFound', 'SimplePath', 'distribution', 'distributions', @@ -70,6 +71,10 @@ def name(self) -> str: # type: ignore[override] # make readonly return name +class MetadataNotFound(FileNotFoundError): + """No metadata file is present in the distribution.""" + + class Sectioned: """ A simple entry point config parser for performance @@ -340,7 +345,7 @@ def select(self, **params) -> EntryPoints: Select entry points from self that match the given parameters (typically group and/or name). """ - return EntryPoints(ep for ep in self if py39.ep_matches(ep, **params)) + return EntryPoints(ep for ep in self if ep.matches(**params)) @property def names(self) -> set[str]: @@ -491,7 +496,12 @@ def _prefer_valid(dists: Iterable[Distribution]) -> Iterable[Distribution]: Ref python/importlib_resources#489. """ - buckets = bucket(dists, lambda dist: bool(dist.metadata)) + + has_metadata = ExceptionTrap(MetadataNotFound).passes( + operator.attrgetter('metadata') + ) + + buckets = bucket(dists, has_metadata) return itertools.chain(buckets[True], buckets[False]) @staticmethod @@ -512,7 +522,7 @@ def _discover_resolvers(): return filter(None, declared) @property - def metadata(self) -> _meta.PackageMetadata | None: + def metadata(self) -> _meta.PackageMetadata: """Return the parsed metadata for this Distribution. The returned object will have keys that name the various bits of @@ -521,6 +531,8 @@ def metadata(self) -> _meta.PackageMetadata | None: Custom providers may provide the METADATA file or override this property. + + :raises MetadataNotFound: If no metadata file is present. """ text = ( @@ -531,20 +543,25 @@ def metadata(self) -> _meta.PackageMetadata | None: # (which points to the egg-info file) attribute unchanged. or self.read_text('') ) - return self._assemble_message(text) + return self._assemble_message(self._ensure_metadata_present(text)) @staticmethod - @pass_none def _assemble_message(text: str) -> _meta.PackageMetadata: # deferred for performance (python/cpython#109829) from . import _adapters return _adapters.Message(email.message_from_string(text)) + def _ensure_metadata_present(self, text: str | None) -> str: + if text is not None: + return text + + raise MetadataNotFound('No package metadata was found.') + @property def name(self) -> str: """Return the 'Name' metadata for the distribution package.""" - return md_none(self.metadata)['Name'] + return self.metadata['Name'] @property def _normalized_name(self): @@ -554,7 +571,7 @@ def _normalized_name(self): @property def version(self) -> str: """Return the 'Version' metadata for the distribution package.""" - return md_none(self.metadata)['Version'] + return self.metadata['Version'] @property def entry_points(self) -> EntryPoints: @@ -1067,11 +1084,12 @@ def distributions(**kwargs) -> Iterable[Distribution]: return Distribution.discover(**kwargs) -def metadata(distribution_name: str) -> _meta.PackageMetadata | None: +def metadata(distribution_name: str) -> _meta.PackageMetadata: """Get the metadata for the named package. :param distribution_name: The name of the distribution package to query. :return: A PackageMetadata containing the parsed metadata. + :raises MetadataNotFound: If no metadata file is present in the distribution. """ return Distribution.from_name(distribution_name).metadata @@ -1088,7 +1106,7 @@ def version(distribution_name: str) -> str: _unique = functools.partial( unique_everseen, - key=py39.normalized_name, + key=operator.attrgetter('_normalized_name'), ) """ Wrapper for ``distributions`` to return unique distributions by name. @@ -1142,7 +1160,7 @@ def packages_distributions() -> Mapping[str, list[str]]: pkg_to_dist = collections.defaultdict(list) for dist in distributions(): for pkg in _top_level_declared(dist) or _top_level_inferred(dist): - pkg_to_dist[pkg].append(md_none(dist.metadata)['Name']) + pkg_to_dist[pkg].append(dist.metadata['Name']) return dict(pkg_to_dist) diff --git a/pipenv/vendor/importlib_metadata/_context.py b/pipenv/vendor/importlib_metadata/_context.py new file mode 100644 index 0000000000..2635b164ce --- /dev/null +++ b/pipenv/vendor/importlib_metadata/_context.py @@ -0,0 +1,118 @@ +from __future__ import annotations + +import functools +import operator + + +# from jaraco.context 6.1 +class ExceptionTrap: + """ + A context manager that will catch certain exceptions and provide an + indication they occurred. + + >>> with ExceptionTrap() as trap: + ... raise Exception() + >>> bool(trap) + True + + >>> with ExceptionTrap() as trap: + ... pass + >>> bool(trap) + False + + >>> with ExceptionTrap(ValueError) as trap: + ... raise ValueError("1 + 1 is not 3") + >>> bool(trap) + True + >>> trap.value + ValueError('1 + 1 is not 3') + >>> trap.tb + + + >>> with ExceptionTrap(ValueError) as trap: + ... raise Exception() + Traceback (most recent call last): + ... + Exception + + >>> bool(trap) + False + """ + + exc_info = None, None, None + + def __init__(self, exceptions=(Exception,)): + self.exceptions = exceptions + + def __enter__(self): + return self + + @property + def type(self): + return self.exc_info[0] + + @property + def value(self): + return self.exc_info[1] + + @property + def tb(self): + return self.exc_info[2] + + def __exit__(self, *exc_info): + type = exc_info[0] + matches = type and issubclass(type, self.exceptions) + if matches: + self.exc_info = exc_info + return matches + + def __bool__(self): + return bool(self.type) + + def raises(self, func, *, _test=bool): + """ + Wrap func and replace the result with the truth + value of the trap (True if an exception occurred). + + First, give the decorator an alias to support Python 3.8 + Syntax. + + >>> raises = ExceptionTrap(ValueError).raises + + Now decorate a function that always fails. + + >>> @raises + ... def fail(): + ... raise ValueError('failed') + >>> fail() + True + """ + + @functools.wraps(func) + def wrapper(*args, **kwargs): + with ExceptionTrap(self.exceptions) as trap: + func(*args, **kwargs) + return _test(trap) + + return wrapper + + def passes(self, func): + """ + Wrap func and replace the result with the truth + value of the trap (True if no exception). + + First, give the decorator an alias to support Python 3.8 + Syntax. + + >>> passes = ExceptionTrap(ValueError).passes + + Now decorate a function that always fails. + + >>> @passes + ... def fail(): + ... raise ValueError('failed') + + >>> fail() + False + """ + return self.raises(func, _test=operator.not_) diff --git a/pipenv/vendor/importlib_metadata/_functools.py b/pipenv/vendor/importlib_metadata/_functools.py index b1fd04a84a..c159b46e48 100644 --- a/pipenv/vendor/importlib_metadata/_functools.py +++ b/pipenv/vendor/importlib_metadata/_functools.py @@ -1,6 +1,7 @@ import functools import types -from typing import Callable, TypeVar +from collections.abc import Callable +from typing import TypeVar # from jaraco.functools 3.3 diff --git a/pipenv/vendor/importlib_metadata/_typing.py b/pipenv/vendor/importlib_metadata/_typing.py deleted file mode 100644 index 32b1d2b98a..0000000000 --- a/pipenv/vendor/importlib_metadata/_typing.py +++ /dev/null @@ -1,15 +0,0 @@ -import functools -import typing - -from ._meta import PackageMetadata - -md_none = functools.partial(typing.cast, PackageMetadata) -""" -Suppress type errors for optional metadata. - -Although Distribution.metadata can return None when metadata is corrupt -and thus None, allow callers to assume it's not None and crash if -that's the case. - -# python/importlib_metadata#493 -""" diff --git a/pipenv/vendor/importlib_metadata/compat/py39.py b/pipenv/vendor/importlib_metadata/compat/py39.py deleted file mode 100644 index 3eb9c01ecb..0000000000 --- a/pipenv/vendor/importlib_metadata/compat/py39.py +++ /dev/null @@ -1,42 +0,0 @@ -""" -Compatibility layer with Python 3.8/3.9 -""" - -from __future__ import annotations - -from typing import TYPE_CHECKING, Any - -if TYPE_CHECKING: # pragma: no cover - # Prevent circular imports on runtime. - from .. import Distribution, EntryPoint -else: - Distribution = EntryPoint = Any - -from .._typing import md_none - - -def normalized_name(dist: Distribution) -> str | None: - """ - Honor name normalization for distributions that don't provide ``_normalized_name``. - """ - try: - return dist._normalized_name - except AttributeError: - from .. import Prepared # -> delay to prevent circular imports. - - return Prepared.normalize( - getattr(dist, "name", None) or md_none(dist.metadata)['Name'] - ) - - -def ep_matches(ep: EntryPoint, **params) -> bool: - """ - Workaround for ``EntryPoint`` objects without the ``matches`` method. - """ - try: - return ep.matches(**params) - except AttributeError: - from .. import EntryPoint # -> delay to prevent circular imports. - - # Reconstruct the EntryPoint object to make sure it is compatible. - return EntryPoint(ep.name, ep.value, ep.group).matches(**params) diff --git a/pipenv/vendor/pipdeptree/__main__.py b/pipenv/vendor/pipdeptree/__main__.py index 20c9526164..989069fe64 100644 --- a/pipenv/vendor/pipdeptree/__main__.py +++ b/pipenv/vendor/pipdeptree/__main__.py @@ -51,7 +51,7 @@ def main(args: Sequence[str] | None = None) -> int | None: print(f"Failed to query custom interpreter: {e}", file=sys.stderr) # noqa: T201 return 1 - tree = PackageDAG.from_pkgs(pkgs) + tree = PackageDAG.from_pkgs(pkgs, include_extras=options.extras) validate(tree) @@ -79,9 +79,9 @@ def main(args: Sequence[str] | None = None) -> int | None: def _is_text_output(options: Options) -> bool: - if any([options.json, options.json_tree, options.graphviz_format, options.mermaid]): + if any((options.json, options.json_tree, options.graphviz_format, options.mermaid)): return False - return options.output_format in {"freeze", "text"} + return options.output_format in {"freeze", "rich", "text"} def _determine_return_code(warning_printer: WarningPrinter) -> int: diff --git a/pipenv/vendor/pipdeptree/_cli.py b/pipenv/vendor/pipdeptree/_cli.py index 3d290cfdd3..812e1b197b 100644 --- a/pipenv/vendor/pipdeptree/_cli.py +++ b/pipenv/vendor/pipdeptree/_cli.py @@ -27,13 +27,14 @@ class Options(Namespace): mermaid: bool graphviz_format: str | None output_format: str + extras: bool depth: float encoding: str license: bool # NOTE: graphviz-* has been intentionally left out. Users of this var should handle it separately. -ALLOWED_RENDER_FORMATS = ["freeze", "json", "json-tree", "mermaid", "text"] +ALLOWED_RENDER_FORMATS = ["freeze", "json", "json-tree", "mermaid", "rich", "text"] class _Formatter(ArgumentDefaultsHelpFormatter): @@ -89,6 +90,13 @@ def build_parser() -> ArgumentParser: help="used along with --exclude to also exclude dependencies of packages", action="store_true", ) + select.add_argument( + "-x", + "--extras", + action="store_true", + default=False, + help="include optional (extras) dependencies in the tree", + ) scope = select.add_mutually_exclusive_group() scope.add_argument( @@ -114,14 +122,14 @@ def build_parser() -> ArgumentParser: metavar="E", ) render.add_argument( - "-a", "--all", action="store_true", help="list all deps at top level (text and freeze render only)" + "-a", "--all", action="store_true", help="list all deps at top level (text, rich, and freeze render only)" ) render.add_argument( "-d", "--depth", type=lambda x: int(x) if x.isdigit() and (int(x) >= 0) else parser.error("Depth must be a number that is >= 0"), default=float("inf"), - help="limit the depth of the tree (text, freeze, and graphviz render only)", + help="limit the depth of the tree (text, rich, freeze, and graphviz render only)", metavar="D", ) render.add_argument( @@ -137,7 +145,7 @@ def build_parser() -> ArgumentParser: render.add_argument( "--license", action="store_true", - help="list the license(s) of a package (text render only)", + help="list the license(s) of a package (text and rich render only)", ) render_type = render.add_mutually_exclusive_group() diff --git a/pipenv/vendor/pipdeptree/_detect_env.py b/pipenv/vendor/pipdeptree/_detect_env.py index 45a70d23f8..85fca70cc6 100644 --- a/pipenv/vendor/pipdeptree/_detect_env.py +++ b/pipenv/vendor/pipdeptree/_detect_env.py @@ -27,7 +27,7 @@ def detect_active_interpreter() -> str: if not path: continue if not path.exists(): - break + continue return str(path) print("Unable to detect virtual environment.", file=sys.stderr) # noqa: T201 diff --git a/pipenv/vendor/pipdeptree/_discovery.py b/pipenv/vendor/pipdeptree/_discovery.py index 4b0b8c0696..8cca6ed1ca 100644 --- a/pipenv/vendor/pipdeptree/_discovery.py +++ b/pipenv/vendor/pipdeptree/_discovery.py @@ -65,7 +65,7 @@ def query_interpreter_for_paths(interpreter: str, *, local_only: bool = False) - args = [interpreter, "-c", cmd] try: result = subprocess.run(args, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, check=True, text=True) # noqa: S603 - return ast.literal_eval(result.stdout) # type: ignore[no-any-return] + return ast.literal_eval(result.stdout) except Exception as e: raise InterpreterQueryError(str(e)) from e @@ -116,7 +116,10 @@ def filter_valid_distributions(iterable_dists: Iterable[Distribution]) -> list[D def has_valid_metadata(dist: Distribution) -> bool: - return "Name" in dist.metadata + try: + return "Name" in dist.metadata + except (TypeError, FileNotFoundError): + return False def render_invalid_metadata_text(site_dirs_with_invalid_metadata: set[str]) -> None: diff --git a/pipenv/vendor/pipdeptree/_freeze.py b/pipenv/vendor/pipdeptree/_freeze.py deleted file mode 100644 index a741791c71..0000000000 --- a/pipenv/vendor/pipdeptree/_freeze.py +++ /dev/null @@ -1,89 +0,0 @@ -from __future__ import annotations - -import locale -from json import JSONDecodeError -from pathlib import Path -from typing import TYPE_CHECKING, Any - -from pipenv.patched.pip._internal.models.direct_url import ( # noqa: PLC2701 - DirectUrl, - DirectUrlValidationError, -) -from pipenv.patched.pip._internal.utils.egg_link import egg_link_path_from_sys_path # noqa: PLC2701 -from pipenv.patched.pip._vendor.packaging.version import Version # noqa: PLC2701 - -if TYPE_CHECKING: - from importlib.metadata import Distribution - - -def dist_to_frozen_repr(dist: Distribution) -> str: - """Return the frozen requirement repr of a `importlib.metadata.Distribution` object.""" - from pipenv.patched.pip._internal.operations.freeze import FrozenRequirement # noqa: PLC0415, PLC2701 - - adapter = PipBaseDistributionAdapter(dist) - fr = FrozenRequirement.from_dist(adapter) # type: ignore[arg-type] - - return str(fr).strip() - - -class PipBaseDistributionAdapter: - """ - An adapter class for pip's `pipenv.patched.pip._internal.metadata.BaseDistribution` abstract class. - - It essentially wraps over an importlib.metadata.Distribution object and provides just enough fields/methods found in - pip's `BaseDistribution` so that we can use `pipenv.patched.pip._internal.operations.freeze.FrozenRequirement.from_dist()`. - - :param dist: Represents an `importlib.metadata.Distribution` object. - """ - - DIRECT_URL_METADATA_NAME = "direct_url.json" - - def __init__(self, dist: Distribution) -> None: - self._dist = dist - self._raw_name = dist.metadata["Name"] - self._version = Version(dist.version) - - @property - def raw_name(self) -> str | Any: - return self._raw_name - - @property - def version(self) -> Version: - return self._version - - @property - def editable(self) -> bool: - return self.editable_project_location is not None - - @property - def direct_url(self) -> DirectUrl | None: - result = None - json_str = self._dist.read_text(self.DIRECT_URL_METADATA_NAME) - try: - if json_str: - result = DirectUrl.from_json(json_str) - except ( - UnicodeDecodeError, - JSONDecodeError, - DirectUrlValidationError, - ): - return result - return result - - @property - def editable_project_location(self) -> str | None: - direct_url = self.direct_url - if direct_url and direct_url.is_local_editable(): - from pipenv.patched.pip._internal.utils.urls import url_to_path # noqa: PLC2701, PLC0415 - - return url_to_path(direct_url.url) - - result = None - egg_link_path = egg_link_path_from_sys_path(self.raw_name) - if egg_link_path: - with Path(egg_link_path).open("r", encoding=locale.getpreferredencoding(False)) as f: # noqa: FBT003 - result = f.readline().rstrip() - return result - - -__all__ = ["dist_to_frozen_repr"] diff --git a/pipenv/vendor/pipdeptree/_models/dag.py b/pipenv/vendor/pipdeptree/_models/dag.py index 357c27c27a..d20acb720d 100644 --- a/pipenv/vendor/pipdeptree/_models/dag.py +++ b/pipenv/vendor/pipdeptree/_models/dag.py @@ -26,14 +26,6 @@ class IncludePatternNotFoundError(Exception): """Include patterns weren't found when filtering a `PackageDAG`.""" -def render_invalid_reqs_text(dist_name_to_invalid_reqs_dict: dict[str, list[str]]) -> None: - for dist_name, invalid_reqs in dist_name_to_invalid_reqs_dict.items(): - print(dist_name, file=sys.stderr) # noqa: T201 - - for invalid_req in invalid_reqs: - print(f' Skipping "{invalid_req}"', file=sys.stderr) # noqa: T201 - - class PackageDAG(Mapping[DistPackage, list[ReqPackage]]): """ Representation of Package dependencies as directed acyclic graph using a dict as the underlying datastructure. @@ -58,33 +50,36 @@ class PackageDAG(Mapping[DistPackage, list[ReqPackage]]): """ @classmethod - def from_pkgs(cls, pkgs: list[Distribution]) -> PackageDAG: + def from_pkgs( + cls, + pkgs: list[Distribution], + *, + include_extras: bool = False, + ) -> PackageDAG: warning_printer = get_warning_printer() dist_pkgs = [DistPackage(p) for p in pkgs] idx = {p.key: p for p in dist_pkgs} - m: dict[DistPackage, list[ReqPackage]] = {} + pkg_deps: dict[DistPackage, list[ReqPackage]] = {} dist_name_to_invalid_reqs_dict: dict[str, list[str]] = {} - for p in dist_pkgs: - reqs = [] - requires_iterator = p.requires() + for pkg in dist_pkgs: + reqs: list[ReqPackage] = [] + requires_iterator = pkg.requires() while True: try: req = next(requires_iterator) except InvalidRequirementError as err: - # We can't work with invalid requirement strings. Let's warn the user about them. if warning_printer.should_warn(): - dist_name_to_invalid_reqs_dict.setdefault(p.project_name, []).append(str(err)) + dist_name_to_invalid_reqs_dict.setdefault(pkg.project_name, []).append(str(err)) continue except StopIteration: break - d = idx.get(canonicalize_name(req.name)) + dist = idx.get(canonicalize_name(req.name)) # Distribution.requires only returns the name of requirements in the metadata file, which may not be the # same as the name in PyPI. We should try to retain the original package names for requirements. # See https://github.com/tox-dev/pipdeptree/issues/242 - req.name = d.project_name if d is not None else req.name - pkg = ReqPackage(req, d) - reqs.append(pkg) - m[p] = reqs + req.name = dist.project_name if dist is not None else req.name + reqs.append(ReqPackage(req, dist)) + pkg_deps[pkg] = reqs should_print_warning = warning_printer.should_warn() and dist_name_to_invalid_reqs_dict if should_print_warning: @@ -93,7 +88,10 @@ def from_pkgs(cls, pkgs: list[Distribution]) -> PackageDAG: lambda: render_invalid_reqs_text(dist_name_to_invalid_reqs_dict), ) - return cls(m) + if include_extras: + _resolve_extras(pkg_deps, idx) + + return cls(pkg_deps) def __init__(self, m: dict[DistPackage, list[ReqPackage]]) -> None: """ @@ -105,7 +103,7 @@ def __init__(self, m: dict[DistPackage, list[ReqPackage]]) -> None: """ self._obj: dict[DistPackage, list[ReqPackage]] = m - self._index: dict[str, DistPackage] = {p.key: p for p in list(self._obj)} + self._index: dict[str, DistPackage] = {p.key: p for p in self._obj} def get_node_as_parent(self, node_key: str) -> DistPackage | None: """ @@ -176,7 +174,7 @@ def filter_nodes( # noqa: C901, PLR0912 seen = set() matched_includes: set[str] = set() for node in self._obj: - if should_exclude_node(node.key, exclude): + if any(fnmatch(node.key, e) for e in exclude): continue if include is None: stack.append(node) @@ -195,7 +193,7 @@ def filter_nodes( # noqa: C901, PLR0912 # Perform DFS on the explicitly included nodes so that we can also include their dependencies, if applicable while stack: n = stack.pop() - cldn = [c for c in self._obj[n] if not should_exclude_node(c.key, exclude)] + cldn = [c for c in self._obj[n] if not any(fnmatch(c.key, e) for e in exclude)] m[n] = cldn seen.add(n.key) for c in cldn: @@ -229,7 +227,7 @@ def _build_exclusion_set_with_dependencies(self, old_exclude: set[str]) -> set[s # used by fnmatch (or the exclusion may not even exist in the graph) resolved_exclude: set[str] = set() - resolved_exclude.update(node.key for node in self._obj if should_exclude_node(node.key, old_exclude)) + resolved_exclude.update(node.key for node in self._obj if any(fnmatch(node.key, e) for e in old_exclude)) # Find all possible candidate nodes for exclusion using DFS candidates: set[str] = set() @@ -290,18 +288,16 @@ def reverse(self) -> ReversedPackageDAG: :returns: DAG in the reversed form """ - m: defaultdict[ReqPackage, list[DistPackage]] = defaultdict(list) + reversed_dag: dict[ReqPackage, list[DistPackage]] = {} + key_index: dict[str, ReqPackage] = {} child_keys = {r.key for r in chain.from_iterable(self._obj.values())} - for k, vs in self._obj.items(): - for v in vs: - # if v is already added to the dict, then ensure that - # we are using the same object. This check is required - # as we're using array mutation - node: ReqPackage = next((p for p in m if p.key == v.key), v) - m[node].append(k.as_parent_of(v)) - if k.key not in child_keys: - m[k.as_requirement()] = [] - return ReversedPackageDAG(dict(m)) # type: ignore[arg-type] + for parent, deps in self._obj.items(): + for dep in deps: + node = key_index.setdefault(dep.key, dep) + reversed_dag.setdefault(node, []).append(parent.as_parent_of(dep)) + if parent.key not in child_keys: + reversed_dag[parent.as_requirement()] = [] + return ReversedPackageDAG(dict(reversed_dag)) # type: ignore[arg-type] def sort(self) -> PackageDAG: """ @@ -313,8 +309,8 @@ def sort(self) -> PackageDAG: return self.__class__({k: sorted(v) for k, v in sorted(self._obj.items())}) # Methods required by the abstract base class Mapping - def __getitem__(self, arg: DistPackage) -> list[ReqPackage] | None: # type: ignore[override] - return self._obj.get(arg) + def __getitem__(self, arg: DistPackage) -> list[ReqPackage]: + return self._obj[arg] def __iter__(self) -> Iterator[DistPackage]: return self._obj.__iter__() @@ -323,8 +319,12 @@ def __len__(self) -> int: return len(self._obj) -def should_exclude_node(key: str, exclude: set[str]) -> bool: - return any(fnmatch(key, e) for e in exclude) +def render_invalid_reqs_text(dist_name_to_invalid_reqs_dict: dict[str, list[str]]) -> None: + for dist_name, invalid_reqs in dist_name_to_invalid_reqs_dict.items(): + print(dist_name, file=sys.stderr) # noqa: T201 + + for invalid_req in invalid_reqs: + print(f' Skipping "{invalid_req}"', file=sys.stderr) # noqa: T201 class ReversedPackageDAG(PackageDAG): @@ -345,18 +345,90 @@ def reverse(self) -> PackageDAG: # type: ignore[override] :returns: reverse of the reversed DAG """ - m: defaultdict[DistPackage, list[ReqPackage]] = defaultdict(list) + forward_dag: dict[DistPackage, list[ReqPackage]] = {} + key_index: dict[str, DistPackage] = {} child_keys = {r.key for r in chain.from_iterable(self._obj.values())} - for k, vs in self._obj.items(): - for v in vs: - assert isinstance(v, DistPackage) - node = next((p for p in m if p.key == v.key), v.as_parent_of(None)) - m[node].append(k) - if k.key not in child_keys: - assert isinstance(k, ReqPackage) - assert k.dist is not None - m[k.dist] = [] - return PackageDAG(dict(m)) + for req_node, parents in self._obj.items(): + for parent in parents: + assert isinstance(parent, DistPackage) + node = key_index.setdefault(parent.key, parent.as_parent_of(None)) + forward_dag.setdefault(node, []).append(req_node) # type: ignore[invalid-argument-type] # runtime: ReqPackage + if req_node.key not in child_keys: + assert isinstance(req_node, ReqPackage) + assert req_node.dist is not None + forward_dag.setdefault(key_index.setdefault(req_node.dist.key, req_node.dist), []) + return PackageDAG(dict(forward_dag)) + + +def _resolve_extras(pkg_deps: dict[DistPackage, list[ReqPackage]], idx: dict[str, DistPackage]) -> None: + """Add extra/optional dependencies to the DAG in-place.""" + extras_needed = _collect_explicit_extras(pkg_deps) + for pkg_key, extras in _collect_satisfied_extras(pkg_deps, idx).items(): + extras_needed.setdefault(pkg_key, set()).update(extras) + processed: dict[str, set[str]] = {} + while extras_needed: + next_round: dict[str, set[str]] = {} + for pkg_key, extras in extras_needed.items(): + new_extras = extras - processed.get(pkg_key, set()) + if not new_extras: + continue + processed.setdefault(pkg_key, set()).update(new_extras) + dist_pkg = idx.get(pkg_key) + if dist_pkg is None or dist_pkg not in pkg_deps: + continue + for req, extra_name in dist_pkg.requires_for_extras(frozenset(new_extras)): + dist = idx.get(canonicalize_name(req.name)) + if dist is None: + continue + req.name = dist.project_name + pkg_deps[dist_pkg].append(ReqPackage(req, dist, extra=extra_name)) + if req.extras: + next_round.setdefault(dist.key, set()).update(req.extras) + extras_needed = next_round + + +def _collect_explicit_extras(pkg_deps: dict[DistPackage, list[ReqPackage]]) -> dict[str, set[str]]: + """Collect extras explicitly requested via requirement specifiers (e.g. ``oauthlib[signedtoken]``).""" + extras_needed: dict[str, set[str]] = {} + for deps in pkg_deps.values(): + for dep in deps: + if dep._obj.extras: # noqa: SLF001 + extras_needed.setdefault(dep.key, set()).update(dep._obj.extras) # noqa: SLF001 + return extras_needed + + +def _collect_satisfied_extras( + pkg_deps: dict[DistPackage, list[ReqPackage]], idx: dict[str, DistPackage] +) -> dict[str, set[str]]: + """Collect extras whose dependencies are all installed in the environment.""" + extras_needed: dict[str, set[str]] = {} + for dist_pkg in pkg_deps: + for extra_name in dist_pkg.provides_extras: + if _extra_is_satisfied(dist_pkg.key, extra_name, pkg_deps, idx, set()): + extras_needed.setdefault(dist_pkg.key, set()).add(extra_name) + return extras_needed + + +def _extra_is_satisfied( + pkg_key: str, + extra_name: str, + pkg_deps: dict[DistPackage, list[ReqPackage]], + idx: dict[str, DistPackage], + visited: set[tuple[str, str]], +) -> bool: + if (pkg_key, extra_name) in visited: + return True + visited.add((pkg_key, extra_name)) + if (dist_pkg := idx.get(pkg_key)) is None or dist_pkg not in pkg_deps: + return False + if not (reqs := list(dist_pkg.requires_for_extras(frozenset({extra_name})))): + return False + for req, _ in reqs: + if (dep_key := canonicalize_name(req.name)) not in idx: + return False + if not all(_extra_is_satisfied(dep_key, sub_extra, pkg_deps, idx, visited) for sub_extra in req.extras): + return False + return True __all__ = [ diff --git a/pipenv/vendor/pipdeptree/_models/package.py b/pipenv/vendor/pipdeptree/_models/package.py index a1de654c8c..ae56d50524 100644 --- a/pipenv/vendor/pipdeptree/_models/package.py +++ b/pipenv/vendor/pipdeptree/_models/package.py @@ -9,7 +9,7 @@ from pipenv.vendor.packaging.requirements import InvalidRequirement, Requirement from pipenv.vendor.packaging.utils import canonicalize_name -from pipenv.vendor.pipdeptree._freeze import dist_to_frozen_repr +from pipenv.vendor.pipdeptree._parser import distribution_to_specifier if TYPE_CHECKING: from collections.abc import Iterator @@ -50,7 +50,7 @@ def licenses(self) -> str: license_str = line.rsplit(":: ", 1)[-1] license_strs.append(license_str) - if len(license_strs) == 0: + if not license_strs: return self.UNKNOWN_LICENSE_STR return f"({', '.join(license_strs)})" @@ -77,8 +77,8 @@ def render( return render(frozen=frozen) @staticmethod - def as_frozen_repr(dist: Distribution) -> str: - return dist_to_frozen_repr(dist) + def as_frozen_repr(distribution: Distribution) -> str: + return distribution_to_specifier(distribution) def __repr__(self) -> str: return f'<{self.__class__.__name__}("{self.key}")>' @@ -120,6 +120,24 @@ def requires(self) -> Iterator[Requirement]: # only required dependencies. yield req + @property + def provides_extras(self) -> frozenset[str]: + return frozenset(self._obj.metadata.get_all("Provides-Extra") or ()) + + def requires_for_extras(self, extras: frozenset[str]) -> Iterator[tuple[Requirement, str]]: + """Yield (requirement, extra_name) for requirements gated behind the given extras.""" + for raw_req in self._obj.requires or []: + try: + req = Requirement(raw_req) + except InvalidRequirement: + continue + if not req.marker or req.marker.evaluate(): + continue + for extra in extras: + if req.marker.evaluate({"extra": extra}): + yield req, extra + break + @property def version(self) -> str: return self._obj.version @@ -138,7 +156,8 @@ def render_as_branch(self, *, frozen: bool) -> str: parent_str = self.req.project_name if parent_ver_spec: parent_str += parent_ver_spec - return f"{self.project_name}=={self.version} [requires: {parent_str}]" + extra_str = f", extra: {self.req.extra}" if self.req.extra else "" + return f"{self.project_name}=={self.version} [requires: {parent_str}{extra_str}]" return self.render_as_root(frozen=frozen) def as_requirement(self) -> ReqPackage: @@ -162,6 +181,13 @@ def as_parent_of(self, req: ReqPackage | None) -> DistPackage: return self return self.__class__(self._obj, req) + @property + def edge_label(self) -> str: + version = (self.req.version_spec if self.req is not None else None) or "any" + if self.req is not None and self.req.extra: + return f"[{self.req.extra}] {version}" + return version + def as_dict(self) -> dict[str, str]: return {"key": self.key, "package_name": self.project_name, "installed_version": self.version} @@ -177,10 +203,11 @@ class ReqPackage(Package): UNKNOWN_VERSION = "?" - def __init__(self, obj: Requirement, dist: DistPackage | None = None) -> None: + def __init__(self, obj: Requirement, dist: DistPackage | None = None, extra: str | None = None) -> None: super().__init__(obj.name) self._obj = obj self.dist = dist + self.extra = extra def render_as_root(self, *, frozen: bool) -> str: if not frozen: @@ -192,16 +219,21 @@ def render_as_root(self, *, frozen: bool) -> str: def render_as_branch(self, *, frozen: bool) -> str: if not frozen: req_ver = self.version_spec or "Any" - return f"{self.project_name} [required: {req_ver}, installed: {self.installed_version}]" + extra_str = f", extra: {self.extra}" if self.extra else "" + return f"{self.project_name} [required: {req_ver}, installed: {self.installed_version}{extra_str}]" return self.render_as_root(frozen=frozen) @property def version_spec(self) -> str | None: - result = None - specs = sorted(map(str, self._obj.specifier), reverse=True) # `reverse` makes '>' prior to '<' - if specs: - result = ",".join(specs) - return result + specs = sorted(map(str, self._obj.specifier), reverse=True) # type: ignore[invalid-argument-type] # `reverse` makes '>' prior to '<' + return ",".join(specs) if specs else None + + @property + def edge_label(self) -> str: + version = self.version_spec or "any" + if self.extra: + return f"[{self.extra}] {version}" + return version @property def installed_version(self) -> str: @@ -237,12 +269,15 @@ def is_missing(self) -> bool: return self.installed_version == self.UNKNOWN_VERSION def as_dict(self) -> dict[str, str]: - return { + result = { "key": self.key, "package_name": self.project_name, "installed_version": self.installed_version, "required_version": self.version_spec if self.version_spec is not None else "Any", } + if self.extra: + result["extra"] = self.extra + return result __all__ = [ diff --git a/pipenv/vendor/pipdeptree/_parser/__init__.py b/pipenv/vendor/pipdeptree/_parser/__init__.py new file mode 100644 index 0000000000..f4bff7c353 --- /dev/null +++ b/pipenv/vendor/pipdeptree/_parser/__init__.py @@ -0,0 +1,7 @@ +from __future__ import annotations + +from .format import distribution_to_specifier + +__all__ = [ + "distribution_to_specifier", +] diff --git a/pipenv/vendor/pipdeptree/_parser/direct_url.py b/pipenv/vendor/pipdeptree/_parser/direct_url.py new file mode 100644 index 0000000000..679fbe11f7 --- /dev/null +++ b/pipenv/vendor/pipdeptree/_parser/direct_url.py @@ -0,0 +1,248 @@ +from __future__ import annotations + +import json +import re +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Final + +if TYPE_CHECKING: + from importlib.metadata import Distribution + + +def get_direct_url(distribution: Distribution) -> DirectUrl | None: + """ + Read and parse direct_url.json from a distribution's metadata. + + :param distribution: Distribution to read direct_url.json from + :returns: Parsed DirectUrl object, or None if file doesn't exist or cannot be parsed + """ + try: + return parse_direct_url_json(json_str) if (json_str := distribution.read_text("direct_url.json")) else None + except (UnicodeDecodeError, DirectUrlValidationError): + return None + + +def parse_direct_url_json(json_str: str) -> DirectUrl: + """ + Parse direct_url.json content into structured DirectUrl object. + + :param json_str: JSON string content from direct_url.json metadata file + :returns: Parsed DirectUrl object + :raises DirectUrlValidationError: If JSON is malformed or missing required fields + """ + data = _load_and_validate_json(json_str) + _validate_required_fields(data) + info = _parse_info_block(data) + return DirectUrl( + url=data["url"], + info=info, + subdirectory=data.get("subdirectory"), + ) + + +def _load_and_validate_json(json_str: str) -> dict: + """Load and validate JSON structure.""" + try: + data = json.loads(json_str) + except json.JSONDecodeError as e: + msg = f"Invalid JSON: {e}" + raise DirectUrlValidationError(msg) from e + if not isinstance(data, dict): + msg = "direct_url.json must be a JSON object" + raise DirectUrlValidationError(msg) + return data + + +def _validate_required_fields(data: dict) -> None: + """Validate required fields and their types.""" + if "url" not in data: + msg = "Missing required 'url' field" + raise DirectUrlValidationError(msg) + if not isinstance(data["url"], str): + msg = "url must be a string" + raise DirectUrlValidationError(msg) + subdirectory = data.get("subdirectory") + if subdirectory is not None and not isinstance(subdirectory, str): + msg = "subdirectory must be a string" + raise DirectUrlValidationError(msg) + + +def _parse_info_block(data: dict) -> VcsInfo | ArchiveInfo | DirInfo: + """Parse and validate info block (vcs_info, archive_info, or dir_info).""" + infos = [] + for key, parser in [ + ("vcs_info", _parse_vcs_info), + ("archive_info", _parse_archive_info), + ("dir_info", _parse_dir_info), + ]: + if key in data: + if not isinstance(data[key], dict): + msg = f"{key} must be a dict" + raise DirectUrlValidationError(msg) + infos.append(parser(data[key])) + if not infos: + msg = "Missing one of vcs_info, archive_info, or dir_info" + raise DirectUrlValidationError(msg) + if len(infos) > 1: + msg = "More than one of vcs_info, archive_info, or dir_info specified" + raise DirectUrlValidationError(msg) + return infos[0] + + +def _parse_vcs_info(vcs_data: dict) -> VcsInfo: + """Parse vcs_info dictionary into VcsInfo object.""" + if "vcs" not in vcs_data: + msg = "Missing required vcs_info.vcs field" + raise DirectUrlValidationError(msg) + if not isinstance(vcs_data["vcs"], str): + msg = "vcs_info.vcs must be a string" + raise DirectUrlValidationError(msg) + commit_id = vcs_data.get("commit_id") + if commit_id is not None and not isinstance(commit_id, str): + msg = "vcs_info.commit_id must be a string" + raise DirectUrlValidationError(msg) + requested_revision = vcs_data.get("requested_revision") + if requested_revision is not None and not isinstance(requested_revision, str): + msg = "vcs_info.requested_revision must be a string" + raise DirectUrlValidationError(msg) + return VcsInfo( + vcs=vcs_data["vcs"], + commit_id=commit_id, + requested_revision=requested_revision, + ) + + +def _parse_archive_info(archive_data: dict) -> ArchiveInfo: + """Parse archive_info dictionary into ArchiveInfo object.""" + hash_value = archive_data.get("hash") + if hash_value is not None: + if not isinstance(hash_value, str): + msg = "archive_info.hash must be a string" + raise DirectUrlValidationError(msg) + if "=" not in hash_value or len(hash_value.split("=", 1)) != 2: + msg = f"invalid archive_info.hash format: {hash_value!r}" + raise DirectUrlValidationError(msg) + hashes: dict[str, str] = {} + if raw_hashes := archive_data.get("hashes"): + if not isinstance(raw_hashes, dict): + msg = "archive_info.hashes must be a dict" + raise DirectUrlValidationError(msg) + hashes = {str(k): str(v) for k, v in raw_hashes.items()} + if not hashes and hash_value: + algo, digest = hash_value.split("=", 1) + hashes = {algo: digest} + return ArchiveInfo(hash=hash_value, hashes=hashes) + + +def _parse_dir_info(dir_data: dict) -> DirInfo: + """Parse dir_info dictionary into DirInfo object.""" + editable = dir_data.get("editable", False) + if not isinstance(editable, bool): + msg = "dir_info.editable must be a boolean" + raise DirectUrlValidationError(msg) + return DirInfo(editable=editable) + + +class DirectUrlValidationError(ValueError): + """Raised when direct_url.json has invalid structure or missing required fields.""" + + +@dataclass +class DirectUrl: + """ + PEP 610 direct_url.json metadata representation. + + See: https://peps.python.org/pep-0610/ + """ + + url: str + info: VcsInfo | ArchiveInfo | DirInfo + subdirectory: str | None = None + + def is_editable(self) -> bool: + """Check if this is an editable installation.""" + return isinstance(self.info, DirInfo) and self.info.editable + + @property + def redacted_url(self) -> str: + """URL with credentials removed, preserving git@ for git VCS and ${VAR} patterns.""" + return _redact_url(self.url, self.info) + + +@dataclass +class VcsInfo: + """ + Version control system information from direct_url.json. + + See PEP 610: https://peps.python.org/pep-0610/ + """ + + vcs: str + commit_id: str | None = None + requested_revision: str | None = None + + +@dataclass +class ArchiveInfo: + """ + Archive information from direct_url.json. + + See PEP 610: https://peps.python.org/pep-0610/ + """ + + hash: str | None = None + hashes: dict[str, str] = field(default_factory=dict) + + +@dataclass +class DirInfo: + """ + Directory information from direct_url.json. + + See PEP 610: https://peps.python.org/pep-0610/ + """ + + editable: bool = False + + +_CREDENTIAL_RE: Final[re.Pattern[str]] = re.compile( + r""" + (?P[a-z+]+://) # URL scheme, e.g. 'https://' + (?P[^@]+) # userinfo before @ + @ # literal @ + """, + re.IGNORECASE | re.VERBOSE, +) +_ENV_VAR_RE: Final[re.Pattern[str]] = re.compile( + r""" + ^ + \$\{[A-Za-z0-9_-]+\} # ${VAR} + (?: : \$\{[A-Za-z0-9_-]+\} )? # optional :${VAR} for ${USER}:${PASS} + $ + """, + re.VERBOSE, +) + + +def _redact_url(url: str, info: VcsInfo | ArchiveInfo | DirInfo) -> str: + """Strip credentials from URL, preserving git@ for git VCS and ${VAR} patterns.""" + match = _CREDENTIAL_RE.match(url) + if not match: + return url + userinfo = match.group("userinfo") + if isinstance(info, VcsInfo) and info.vcs == "git" and userinfo == "git": + return url + if _ENV_VAR_RE.match(userinfo): + return url + return f"{match.group('scheme')}{url[match.end() :]}" + + +__all__ = [ + "ArchiveInfo", + "DirInfo", + "DirectUrl", + "DirectUrlValidationError", + "VcsInfo", + "get_direct_url", + "parse_direct_url_json", +] diff --git a/pipenv/vendor/pipdeptree/_parser/editable.py b/pipenv/vendor/pipdeptree/_parser/editable.py new file mode 100644 index 0000000000..af3a6a0bc2 --- /dev/null +++ b/pipenv/vendor/pipdeptree/_parser/editable.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +import locale +import os +import re +import site +import string +import sys +from pathlib import Path +from typing import TYPE_CHECKING +from urllib.parse import urlsplit +from urllib.request import url2pathname + +from .direct_url import get_direct_url + +if TYPE_CHECKING: + from importlib.metadata import Distribution + + +def get_editable_location(distribution: Distribution) -> str | None: + """ + Get source location for an editable installation. + + Tries modern PEP 610 direct_url.json first, then falls back to legacy .egg-link files. + See: https://peps.python.org/pep-0610/ + + :param distribution: Distribution to check + :returns: Path to editable source location, or None if package is not editable + """ + if (direct_url := get_direct_url(distribution)) and direct_url.is_editable(): + return url_to_path(direct_url.url) + if egg_link := find_egg_link(distribution.metadata["Name"]): + return read_egg_link_location(egg_link) + return None + + +def url_to_path(url: str) -> str: + """ + Convert file:// URL to filesystem path. + + Handles localhost URLs, UNC paths on Windows, and Windows drive letter corrections. + Matches pip's url_to_path implementation for compatibility. + + :param url: URL to convert (must have file:// scheme) + :returns: Filesystem path + :raises ValueError: If URL doesn't use file:// scheme or is non-local on non-Windows platforms + """ + if not url.startswith("file:"): + msg = f"You can only turn file: urls into filenames (not {url!r})" + raise ValueError(msg) + _, netloc, path, _, _ = urlsplit(url) + if not netloc or netloc == "localhost": + netloc = "" + elif os.name == "nt": + netloc = "\\\\" + netloc # pragma: win32 cover + else: + msg = f"non-local file URIs are not supported on this platform: {url!r}" + raise ValueError(msg) + path = url2pathname(netloc + path) + if ( + os.name == "nt" # noqa: PLR0916 + and not netloc + and len(path) >= 3 + and path[0] == "/" + and path[1] in string.ascii_letters + and path[2:4] in {":", ":/"} + ): + path = path[1:] # pragma: win32 cover + return path + + +def find_egg_link(package_name: str) -> Path | None: + """ + Find .egg-link file for legacy editable installations. + + Matches pip's egg_link_path_from_sys_path: searches sys.path entries first, + then falls back to site-packages and user site-packages. + + :param package_name: Name of package to search for + :returns: Path to .egg-link file if found, None otherwise + """ + candidates = _egg_link_names(package_name) + for search_dir in sys.path: + for name in candidates: + if (egg_link := Path(search_dir) / name).is_file(): + return egg_link + site_dirs = site.getsitepackages() if hasattr(site, "getsitepackages") else [] + if user_site := site.getusersitepackages(): + site_dirs.append(user_site) + for site_dir in site_dirs: + for name in candidates: + if (egg_link := Path(site_dir) / name).is_file(): + return egg_link + return None + + +def _egg_link_names(package_name: str) -> list[str]: + """Generate candidate egg-link filenames: safe-name normalized and raw.""" + safe_name = re.sub(r"[^A-Za-z0-9.]+", "-", package_name) + candidates = [f"{safe_name}.egg-link"] + if safe_name != package_name: + candidates.append(f"{package_name}.egg-link") + return candidates + + +def read_egg_link_location(egg_link_path: Path) -> str: + """ + Read source directory path from .egg-link file. + + The first line of an .egg-link file contains the absolute path to the source directory. + + :param egg_link_path: Path to .egg-link file + :returns: Source directory path + """ + with egg_link_path.open("r", encoding=locale.getpreferredencoding(do_setlocale=False)) as f: + return f.readline().rstrip() + + +__all__ = [ + "find_egg_link", + "get_editable_location", + "read_egg_link_location", + "url_to_path", +] diff --git a/pipenv/vendor/pipdeptree/_parser/format.py b/pipenv/vendor/pipdeptree/_parser/format.py new file mode 100644 index 0000000000..96647ffd56 --- /dev/null +++ b/pipenv/vendor/pipdeptree/_parser/format.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from pipenv.vendor.packaging.version import InvalidVersion, Version + +from .direct_url import ArchiveInfo, VcsInfo, get_direct_url +from .editable import find_egg_link, read_egg_link_location, url_to_path +from .vcs import VcsError, VcsResult, get_vcs_requirement + +if TYPE_CHECKING: + from importlib.metadata import Distribution + + from .direct_url import DirectUrl + + +def distribution_to_specifier(distribution: Distribution) -> str: + """ + Convert distribution to requirement specifier string. + + Handles regular packages (PEP 440 version specifiers), editable installs (PEP 610 direct_url.json and legacy + .egg-link), and direct URL installs (PEP 440 direct references for VCS, archive, directory). + + For editable installs, probes filesystem for VCS information (git/hg/svn/bzr) to generate full VCS URL with commit + hash, falling back to local path with diagnostic comment if VCS not detected or extraction fails. + + See: + - PEP 440: https://peps.python.org/pep-0440/ + - PEP 610: https://peps.python.org/pep-0610/ + + :param distribution: Distribution to convert + :returns: Requirement specifier string + """ + direct_url = get_direct_url(distribution) + if direct_url: + if direct_url.is_editable(): + location = url_to_path(direct_url.url) + return _format_editable(location, distribution.metadata["Name"], distribution.version) + return format_requirement(direct_url, distribution.metadata["Name"]) + if egg_link := find_egg_link(distribution.metadata["Name"]): + location = read_egg_link_location(egg_link) + return _format_editable(location, distribution.metadata["Name"], distribution.version) + name = distribution.metadata["Name"] + try: + Version(distribution.version) + except InvalidVersion: + return f"{name}==={distribution.version}" + else: + return f"{name}=={distribution.version}" + + +def _format_editable(location: str, package_name: str, version: str) -> str: + """ + Format editable install requirement with VCS detection and diagnostic comments. + + Probes location for VCS (git/hg/svn/bzr) and generates VCS URL if detected, + otherwise uses local path with pip-compatible diagnostic comment. + """ + vcs_result = get_vcs_requirement(location, package_name) + if vcs_result.requirement: + return f"-e {vcs_result.requirement}" + comment = _diagnostic_comment(vcs_result, package_name, version) + if comment: + return f"{comment}\n-e {location}" + return f"-e {location}" + + +_DIAGNOSTIC_TEMPLATES: dict[VcsError, str] = { + VcsError.NO_VCS: "# Editable install with no version control ({package_name}=={version})", + VcsError.NO_REMOTE: "# Editable {vcs_name} install with no remote ({package_name}=={version})", + VcsError.INVALID_REMOTE: ( + "# Editable {vcs_name} install ({package_name}=={version}) with either a deleted local remote or invalid URI:" + ), +} + + +def _diagnostic_comment(vcs_result: VcsResult, package_name: str, version: str) -> str | None: + """Generate pip-compatible diagnostic comment based on VcsResult error.""" + if template := _DIAGNOSTIC_TEMPLATES.get(vcs_result.error): + return template.format(package_name=package_name, version=version, vcs_name=vcs_result.vcs_name) + return None + + +def format_requirement(direct_url: DirectUrl, package_name: str) -> str: + """ + Format DirectUrl as PEP 440 direct reference requirement. + + Uses redacted_url to strip credentials from URLs. + + :param direct_url: DirectUrl object to format (from PEP 610 direct_url.json) + :param package_name: Package name to include in requirement string + :returns: Formatted requirement string + """ + url = direct_url.redacted_url + requirement = f"{package_name} @ " + if isinstance(direct_url.info, VcsInfo): + requirement += f"{direct_url.info.vcs}+{url}" + if direct_url.info.commit_id: + requirement += f"@{direct_url.info.commit_id}" + elif isinstance(direct_url.info, ArchiveInfo): + requirement += url + if direct_url.info.hash: + requirement += f"#{direct_url.info.hash}" + else: + requirement += url + if direct_url.subdirectory: + requirement += f"{'&' if '#' in requirement else '#'}subdirectory={direct_url.subdirectory}" + return requirement + + +__all__ = [ + "distribution_to_specifier", + "format_requirement", +] diff --git a/pipenv/vendor/pipdeptree/_parser/vcs/__init__.py b/pipenv/vendor/pipdeptree/_parser/vcs/__init__.py new file mode 100644 index 0000000000..0e2e07ecf2 --- /dev/null +++ b/pipenv/vendor/pipdeptree/_parser/vcs/__init__.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +from .bzr import get_bzr_requirement +from .git import get_git_repo_root, get_git_requirement +from .hg import get_hg_requirement +from .shared import VcsError, VcsResult +from .svn import get_svn_requirement + +if TYPE_CHECKING: + from collections.abc import Callable + + +def get_vcs_requirement(location: str, package_name: str) -> VcsResult: + """ + Detect VCS and generate requirement string for editable install. + + Probes directory for git/hg/svn/bzr repositories (including parent directories), + picks the innermost repo root (matching pip's get_backend_for_dir logic). + + :param location: Filesystem path to source directory + :param package_name: Package name for egg fragment + :returns: VcsResult with requirement string and diagnostic info + """ + roots: dict[str, Callable[[str, str, str], VcsResult]] = {} + if git_root := get_git_repo_root(location): + roots[git_root] = get_git_requirement + if hg_root := _find_marker_root(location, ".hg", dir_only=False): + roots[hg_root] = get_hg_requirement + if svn_root := _find_marker_root(location, ".svn"): + roots[svn_root] = get_svn_requirement + if bzr_root := _find_marker_root(location, ".bzr"): + roots[bzr_root] = get_bzr_requirement + if not roots: + return VcsResult(None, error=VcsError.NO_VCS) + innermost = "" + for root in roots: + if len(root) > len(innermost): + innermost = root + return roots[innermost](location, package_name, innermost) + + +def _find_marker_root(location: str, marker: str, *, dir_only: bool = True) -> str | None: + """Walk up from location looking for a directory containing marker (e.g. .hg, .svn, .bzr).""" + current = Path(location).resolve() + check = Path.is_dir if dir_only else Path.exists + while True: + if check(current / marker): + return str(current) + parent = current.parent + if parent == current: + return None + current = parent + + +__all__ = [ + "VcsError", + "VcsResult", + "get_vcs_requirement", +] diff --git a/pipenv/vendor/pipdeptree/_parser/vcs/bzr.py b/pipenv/vendor/pipdeptree/_parser/vcs/bzr.py new file mode 100644 index 0000000000..1a9cd7054f --- /dev/null +++ b/pipenv/vendor/pipdeptree/_parser/vcs/bzr.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +import subprocess # noqa: S404 +from pathlib import Path + +from .shared import VcsError, VcsResult, build_vcs_result, is_local_path + + +def get_bzr_requirement(location: str, package_name: str, repo_root: str) -> VcsResult: + try: + remote_url = _get_bzr_remote_url(repo_root) + if remote_url is None: + return VcsResult(None, vcs_name="bzr", error=VcsError.NO_REMOTE) + if is_local_path(remote_url): + remote_url = Path(remote_url).as_uri() + if not (revision := _get_bzr_revision(repo_root)): + return VcsResult(None, vcs_name="bzr", error=VcsError.NO_REMOTE) + except FileNotFoundError: + return VcsResult(None, vcs_name="bzr", error=VcsError.COMMAND_NOT_FOUND) + return build_vcs_result( + vcs_name="bzr", + remote_url=remote_url, + commit_id=revision, + package_name=package_name, + location=location, + repo_root=repo_root, + always_prefix=False, + include_subdirectory=False, + ) + + +def _get_bzr_remote_url(repo_root: str) -> str | None: + """Parse `bzr info` output for checkout/parent branch URL.""" + try: + output = subprocess.run( + ["bzr", "info"], # noqa: S607 + cwd=repo_root, + capture_output=True, + text=True, + check=True, + timeout=5, + ).stdout + except (subprocess.CalledProcessError, subprocess.TimeoutExpired): + return None + for line in output.splitlines(): + stripped = line.strip() + for prefix in ("checkout of branch: ", "parent branch: "): + if stripped.startswith(prefix): + return stripped[len(prefix) :].strip() or None + return None + + +def _get_bzr_revision(repo_root: str) -> str | None: + try: + output = subprocess.run( + ["bzr", "revno"], # noqa: S607 + cwd=repo_root, + capture_output=True, + text=True, + check=True, + timeout=5, + ).stdout.strip() + except (subprocess.CalledProcessError, subprocess.TimeoutExpired): + return None + lines = output.splitlines() + return lines[-1].strip() if lines else None + + +__all__ = [ + "get_bzr_requirement", +] diff --git a/pipenv/vendor/pipdeptree/_parser/vcs/git.py b/pipenv/vendor/pipdeptree/_parser/vcs/git.py new file mode 100644 index 0000000000..132a00e0ec --- /dev/null +++ b/pipenv/vendor/pipdeptree/_parser/vcs/git.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +import os +import re +import subprocess # noqa: S404 +from pathlib import Path +from typing import Final + +from .shared import VcsError, VcsResult, build_vcs_result, is_local_path + +_SANITIZED_GIT_VARS: Final[frozenset[str]] = frozenset({"GIT_DIR", "GIT_WORK_TREE"}) +_SCHEME_RE: Final[re.Pattern[str]] = re.compile(r"\w+://") +_SCP_RE: Final[re.Pattern[str]] = re.compile( + r""" + ^ + (?P\w+@)? # Optional user, e.g. 'git@' + (?P[^/:]+): # Server, e.g. 'github.com' + (?P\w[^:]*) # Server-side path starting with alphanumeric (not Windows C:) + $ + """, + re.VERBOSE, +) + + +def get_git_repo_root(location: str) -> str | None: + """Find git repository root from any subdirectory.""" + try: + repo_root = subprocess.run( + ["git", "rev-parse", "--show-toplevel"], # noqa: S607 + cwd=location, + capture_output=True, + text=True, + check=True, + timeout=5, + env=_git_env(), + ).stdout.strip() + except (subprocess.CalledProcessError, subprocess.TimeoutExpired): + return None + except FileNotFoundError: + return None + else: + return repo_root or None + + +def get_git_requirement(location: str, package_name: str, repo_root: str) -> VcsResult: + try: + remote_url = _get_git_remote_url(repo_root) + if remote_url is None: + return VcsResult(None, vcs_name="git", error=VcsError.NO_REMOTE) + if not (commit_id := _get_git_commit_id(repo_root)): + return VcsResult(None, vcs_name="git", error=VcsError.NO_REMOTE) + except FileNotFoundError: + return VcsResult(None, vcs_name="git", error=VcsError.COMMAND_NOT_FOUND) + normalized = _normalize_git_url(remote_url, repo_root) + if normalized is None: + return VcsResult(None, vcs_name="git", error=VcsError.INVALID_REMOTE) + return build_vcs_result( + vcs_name="git", + remote_url=normalized, + commit_id=commit_id, + package_name=package_name, + location=location, + repo_root=repo_root, + always_prefix=True, + ) + + +def _get_git_remote_url(repo_root: str) -> str | None: + """Get git remote URL, preferring origin.""" + try: + remotes_output = subprocess.run( + ["git", "config", "--get-regexp", r"remote\..*\.url"], # noqa: S607 + cwd=repo_root, + capture_output=True, + text=True, + check=False, + timeout=5, + env=_git_env(), + ).stdout.strip() + if not remotes_output: + return None + remotes = remotes_output.splitlines() + found_remote = remotes[0] + for remote in remotes: + if remote.startswith("remote.origin.url "): + found_remote = remote + break + parts = found_remote.split(" ", 1) + return parts[1].strip() if len(parts) >= 2 and parts[1].strip() else None + except (subprocess.CalledProcessError, subprocess.TimeoutExpired): + return None + + +def _get_git_commit_id(repo_root: str) -> str | None: + """Get current git commit ID.""" + try: + commit_id = subprocess.run( + ["git", "rev-parse", "HEAD"], # noqa: S607 + cwd=repo_root, + capture_output=True, + text=True, + check=True, + timeout=5, + env=_git_env(), + ).stdout.strip() + except (subprocess.CalledProcessError, subprocess.TimeoutExpired): + return None + else: + return commit_id or None + + +def _normalize_git_url(url: str, repo_root: str) -> str | None: + """ + Normalize git URL to standard format matching pip's behavior. + + Returns None for URLs that don't match any known pattern. + """ + if _SCHEME_RE.match(url): + return url + path = Path(url) if is_local_path(url) else Path(repo_root) / url + if path.exists(): + return path.resolve().as_uri() + if match := _SCP_RE.match(url): + return f"ssh://{match.group('user') or ''}{match.group('host')}/{match.group('path')}" + return None + + +def _git_env() -> dict[str, str]: + return {k: v for k, v in os.environ.items() if k not in _SANITIZED_GIT_VARS} + + +__all__ = [ + "get_git_repo_root", + "get_git_requirement", +] diff --git a/pipenv/vendor/pipdeptree/_parser/vcs/hg.py b/pipenv/vendor/pipdeptree/_parser/vcs/hg.py new file mode 100644 index 0000000000..b18e3b01c8 --- /dev/null +++ b/pipenv/vendor/pipdeptree/_parser/vcs/hg.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import subprocess # noqa: S404 +from pathlib import Path + +from .shared import VcsError, VcsResult, build_vcs_result, is_local_path + + +def get_hg_requirement(location: str, package_name: str, repo_root: str) -> VcsResult: + try: + remote_url = _get_hg_remote_url(repo_root) + if remote_url is None: + return VcsResult(None, vcs_name="hg", error=VcsError.NO_REMOTE) + if is_local_path(remote_url): + remote_url = Path(remote_url).as_uri() + if not (commit_id := _get_hg_commit_id(repo_root)): + return VcsResult(None, vcs_name="hg", error=VcsError.NO_REMOTE) + except FileNotFoundError: + return VcsResult(None, vcs_name="hg", error=VcsError.COMMAND_NOT_FOUND) + return build_vcs_result( + vcs_name="hg", + remote_url=remote_url, + commit_id=commit_id, + package_name=package_name, + location=location, + repo_root=repo_root, + always_prefix=False, + ) + + +def _get_hg_remote_url(repo_root: str) -> str | None: + try: + url = subprocess.run( + ["hg", "showconfig", "paths.default"], # noqa: S607 + cwd=repo_root, + capture_output=True, + text=True, + check=True, + timeout=5, + ).stdout.strip() + except (subprocess.CalledProcessError, subprocess.TimeoutExpired): + return None + else: + return url or None + + +def _get_hg_commit_id(repo_root: str) -> str | None: + try: + commit_id = subprocess.run( + ["hg", "parents", "--template={node}"], # noqa: S607 + cwd=repo_root, + capture_output=True, + text=True, + check=True, + timeout=5, + ).stdout.strip() + except (subprocess.CalledProcessError, subprocess.TimeoutExpired): + return None + else: + return commit_id or None + + +__all__ = [ + "get_hg_requirement", +] diff --git a/pipenv/vendor/pipdeptree/_parser/vcs/shared.py b/pipenv/vendor/pipdeptree/_parser/vcs/shared.py new file mode 100644 index 0000000000..718b3661e9 --- /dev/null +++ b/pipenv/vendor/pipdeptree/_parser/vcs/shared.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +import os +import urllib.parse +from dataclasses import dataclass +from enum import Enum, auto +from pathlib import Path + + +class VcsError(Enum): + NONE = auto() + NO_VCS = auto() + NO_REMOTE = auto() + INVALID_REMOTE = auto() + COMMAND_NOT_FOUND = auto() + + +@dataclass +class VcsResult: + requirement: str | None + vcs_name: str | None = None + error: VcsError = VcsError.NONE + + +def build_vcs_result( # noqa: PLR0913, PLR0917 + vcs_name: str, + remote_url: str, + commit_id: str, + package_name: str, + location: str, + repo_root: str, + *, + always_prefix: bool, + include_subdirectory: bool = True, +) -> VcsResult: + """Build VcsResult with requirement string from components.""" + safe_package_name = _normalize_egg_name(package_name) + if always_prefix or not remote_url.lower().startswith(f"{vcs_name}:"): + url = f"{vcs_name}+{remote_url}" + else: + url = remote_url + quoted_commit_id = urllib.parse.quote(commit_id, "/") + result = f"{url}@{quoted_commit_id}#egg={safe_package_name}" + if include_subdirectory and (subdirectory := _find_project_root(location, repo_root)): + result += f"&subdirectory={subdirectory}" + return VcsResult(result, vcs_name=vcs_name) + + +def _normalize_egg_name(name: str) -> str: + return name.replace("-", "_") + + +def _find_project_root(location: str, repo_root: str) -> str | None: + """ + Walk up from location to find the installable project root. + + Matches pip's find_path_to_project_root_from_repo_root: + walks UP from location looking for pyproject.toml or setup.py. + """ + current = Path(location).resolve() + abs_root = Path(repo_root).resolve() + while not _is_installable_dir(current): + parent = Path(current).parent + if parent == current: + return None + current = parent + try: + if Path(abs_root).samefile(current): + return None + except (ValueError, OSError): + return None + return os.path.relpath(current, abs_root) + + +def _is_installable_dir(path: str | Path) -> bool: + resolved = Path(path) + return resolved.is_dir() and ((resolved / "pyproject.toml").is_file() or (resolved / "setup.py").is_file()) + + +def is_local_path(path: str) -> bool: + """Check if path is a local filesystem path (starts with os.sep or has drive letter).""" + if path.startswith(os.sep): + return True + return len(path) >= 2 and path[1] == ":" and path[0].isalpha() + + +__all__ = [ + "VcsError", + "VcsResult", + "build_vcs_result", + "is_local_path", +] diff --git a/pipenv/vendor/pipdeptree/_parser/vcs/svn.py b/pipenv/vendor/pipdeptree/_parser/vcs/svn.py new file mode 100644 index 0000000000..a3d211b3c9 --- /dev/null +++ b/pipenv/vendor/pipdeptree/_parser/vcs/svn.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +import subprocess # noqa: S404 +import xml.etree.ElementTree as ET # noqa: S405 +from pathlib import Path + +from .shared import VcsError, VcsResult, build_vcs_result + + +def get_svn_requirement(location: str, package_name: str, repo_root: str) -> VcsResult: + try: + svn_info = _get_svn_info(location) + except FileNotFoundError: + return VcsResult(None, vcs_name="svn", error=VcsError.COMMAND_NOT_FOUND) + if svn_info is None: + entries = _get_svn_entries_fallback(location) + if entries is None: + return VcsResult(None, vcs_name="svn", error=VcsError.NO_REMOTE) + remote_url, revision = entries + else: + remote_url, revision = svn_info + return build_vcs_result( + vcs_name="svn", + remote_url=remote_url, + commit_id=revision, + package_name=package_name, + location=location, + repo_root=repo_root, + always_prefix=True, + include_subdirectory=False, + ) + + +def _get_svn_info(location: str) -> tuple[str, str] | None: + """Parse svn info --xml to extract URL and revision.""" + try: + xml_output = subprocess.run( + ["svn", "info", "--xml"], # noqa: S607 + cwd=location, + capture_output=True, + text=True, + check=True, + timeout=5, + ).stdout + except (subprocess.CalledProcessError, subprocess.TimeoutExpired): + return None + try: + root = ET.fromstring(xml_output) # noqa: S314 + except ET.ParseError: + return None + entry = root.find(".//entry") + if entry is None: + return None + revision = entry.get("revision", "") + url_elem = entry.find("url") + if url_elem is None or not url_elem.text: + return None + return url_elem.text, revision + + +def _get_svn_entries_fallback(location: str) -> tuple[str, str] | None: + """Parse legacy .svn/entries file (pre-1.7 SVN) for URL and revision.""" + entries_path = Path(location) / ".svn" / "entries" + if not entries_path.is_file(): + return None + try: + data = entries_path.read_text(encoding="utf-8") + except OSError: + return None + if data.startswith(" None: render_mermaid(tree) elif output_format == "freeze": render_freeze(tree, max_depth=options.depth, list_all=options.all) + elif output_format == "rich": + render_rich_text(tree, max_depth=options.depth, list_all=options.all, include_license=options.license) elif output_format.startswith("graphviz-"): render_graphviz( tree, output_format=output_format[len("graphviz-") :], reverse=options.reverse, max_depth=options.depth diff --git a/pipenv/vendor/pipdeptree/_render/freeze.py b/pipenv/vendor/pipdeptree/_render/freeze.py index b8a1c636b9..9ef7805f08 100644 --- a/pipenv/vendor/pipdeptree/_render/freeze.py +++ b/pipenv/vendor/pipdeptree/_render/freeze.py @@ -1,41 +1,16 @@ from __future__ import annotations -from itertools import chain -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING -from .text import get_top_level_nodes +from .text import _render_text_simple, get_top_level_nodes if TYPE_CHECKING: from pipenv.vendor.pipdeptree._models.dag import PackageDAG - from pipenv.vendor.pipdeptree._models.package import DistPackage, ReqPackage def render_freeze(tree: PackageDAG, *, max_depth: float, list_all: bool = True) -> None: nodes = get_top_level_nodes(tree, list_all=list_all) - - def aux( - node: DistPackage | ReqPackage, - parent: DistPackage | ReqPackage | None = None, - indent: int = 0, - cur_chain: list[str] | None = None, - depth: int = 0, - ) -> list[Any]: - cur_chain = cur_chain or [] - node_str = node.render(parent, frozen=True) - if parent: - prefix = " " * indent - node_str = prefix + node_str - result = [node_str] - children = [ - aux(c, node, indent=indent + 2, cur_chain=[*cur_chain, c.project_name], depth=depth + 1) - for c in tree.get_children(node.key) - if c.project_name not in cur_chain and depth + 1 <= max_depth - ] - result += list(chain.from_iterable(children)) - return result - - lines = chain.from_iterable([aux(p) for p in nodes]) - print("\n".join(lines)) # noqa: T201 + _render_text_simple(tree, nodes, max_depth, include_license=False, frozen=True, bullet="") __all__ = [ diff --git a/pipenv/vendor/pipdeptree/_render/graphviz.py b/pipenv/vendor/pipdeptree/_render/graphviz.py index d65b119fc5..942de814e5 100644 --- a/pipenv/vendor/pipdeptree/_render/graphviz.py +++ b/pipenv/vendor/pipdeptree/_render/graphviz.py @@ -14,97 +14,9 @@ from pipenv.vendor.pipdeptree._models import PackageDAG -def _get_all_dep_keys(tree: PackageDAG) -> set[str]: - """Return the set of keys that appear as dependencies of some package.""" - dep_keys: set[str] = set() - for deps in tree.values(): - dep_keys.update(dep.key for dep in deps) - return dep_keys - - -def _build_reverse_graph(tree: PackageDAG, graph: Digraph, max_depth: float) -> None: # noqa: C901, PLR0912 - """Build graphviz nodes and edges for a reversed dependency tree.""" - if max_depth < math.inf: - parent_keys = _get_all_dep_keys(tree) - root_keys = {dep_rev.key for dep_rev in tree if dep_rev.key not in parent_keys} - visited: dict[str, int] = {} - queue: deque[tuple[str, int]] = deque((k, 0) for k in root_keys) - while queue: - key, depth = queue.popleft() - if key in visited: - continue - visited[key] = depth - if depth < max_depth: - for parent in tree.get_children(key): - if parent.key not in visited: - queue.append((parent.key, depth + 1)) - for dep_rev, parents in tree.items(): - if dep_rev.key not in visited: - continue - assert isinstance(dep_rev, ReqPackage) - dep_label = f"{dep_rev.project_name}\\n{dep_rev.installed_version}" - graph.node(dep_rev.key, label=dep_label) - if visited[dep_rev.key] < max_depth: - for parent in parents: - assert isinstance(parent, DistPackage) - if parent.key in visited: - edge_label = (parent.req.version_spec if parent.req is not None else None) or "any" - graph.edge(dep_rev.key, parent.key, label=edge_label) - else: - for dep_rev, parents in tree.items(): - assert isinstance(dep_rev, ReqPackage) - dep_label = f"{dep_rev.project_name}\\n{dep_rev.installed_version}" - graph.node(dep_rev.key, label=dep_label) - for parent in parents: - assert isinstance(parent, DistPackage) - edge_label = (parent.req.version_spec if parent.req is not None else None) or "any" - graph.edge(dep_rev.key, parent.key, label=edge_label) - - -def _build_forward_graph(tree: PackageDAG, graph: Digraph, max_depth: float) -> None: # noqa: C901, PLR0912 - """Build graphviz nodes and edges for a forward dependency tree.""" - if max_depth < math.inf: # noqa: PLR1702 - dep_keys = _get_all_dep_keys(tree) - root_keys = {pkg.key for pkg in tree if pkg.key not in dep_keys} - visited: dict[str, int] = {} - queue: deque[tuple[str, int]] = deque((k, 0) for k in root_keys) - while queue: - key, depth = queue.popleft() - if key in visited: - continue - visited[key] = depth - if depth < max_depth: - children = tree.get_children(key) - for dep in children: - if dep.key not in visited: - queue.append((dep.key, depth + 1)) - for pkg, deps in tree.items(): - if pkg.key not in visited: - continue - pkg_label = f"{pkg.project_name}\\n{pkg.version}" - graph.node(pkg.key, label=pkg_label) - if visited[pkg.key] < max_depth: - for dep in deps: - if dep.key in visited: - edge_label = dep.version_spec or "any" - if dep.is_missing: - dep_label = f"{dep.project_name}\\n(missing)" - graph.node(dep.key, label=dep_label, style="dashed") - graph.edge(pkg.key, dep.key, style="dashed") - else: - graph.edge(pkg.key, dep.key, label=edge_label) - else: - for pkg, deps in tree.items(): - pkg_label = f"{pkg.project_name}\\n{pkg.version}" - graph.node(pkg.key, label=pkg_label) - for dep in deps: - edge_label = dep.version_spec or "any" - if dep.is_missing: - dep_label = f"{dep.project_name}\\n(missing)" - graph.node(dep.key, label=dep_label, style="dashed") - graph.edge(pkg.key, dep.key, style="dashed") - else: - graph.edge(pkg.key, dep.key, label=edge_label) +def render_graphviz(tree: PackageDAG, *, output_format: str, reverse: bool, max_depth: float = math.inf) -> None: + output = dump_graphviz(tree, output_format=output_format, is_reverse=reverse, max_depth=max_depth) + print_graphviz(output) def dump_graphviz( @@ -159,9 +71,9 @@ def dump_graphviz( # As it's unknown if the selected output format is binary or not, try to # decode it as UTF8 and only print it out in binary if that's not possible. try: - return graph.pipe().decode("utf-8") # type: ignore[no-any-return] + return graph.pipe().decode("utf-8") except UnicodeDecodeError: - return graph.pipe() # type: ignore[no-any-return] + return graph.pipe() def print_graphviz(dump_output: str | bytes) -> None: @@ -171,16 +83,71 @@ def print_graphviz(dump_output: str | bytes) -> None: :param dump_output: The output from dump_graphviz """ - if hasattr(dump_output, "encode"): + if isinstance(dump_output, str): print(dump_output) # noqa: T201 else: with os.fdopen(sys.stdout.fileno(), "wb") as bytestream: bytestream.write(dump_output) -def render_graphviz(tree: PackageDAG, *, output_format: str, reverse: bool, max_depth: float = math.inf) -> None: - output = dump_graphviz(tree, output_format=output_format, is_reverse=reverse, max_depth=max_depth) - print_graphviz(output) +def _build_reverse_graph(tree: PackageDAG, graph: Digraph, max_depth: float) -> None: + """Build graphviz nodes and edges for a reversed dependency tree.""" + visited = _compute_reachable_depths(tree, _get_root_keys(tree), max_depth) + + for dep_rev, parents in tree.items(): + if visited is not None and dep_rev.key not in visited: + continue + assert isinstance(dep_rev, ReqPackage) + graph.node(dep_rev.key, label=f"{dep_rev.project_name}\\n{dep_rev.installed_version}") + if visited is None or visited[dep_rev.key] < max_depth: + for parent in parents: + assert isinstance(parent, DistPackage) + if visited is not None and parent.key not in visited: + continue + graph.edge(dep_rev.key, parent.key, label=parent.edge_label) + + +def _build_forward_graph(tree: PackageDAG, graph: Digraph, max_depth: float) -> None: + """Build graphviz nodes and edges for a forward dependency tree.""" + visited = _compute_reachable_depths(tree, _get_root_keys(tree), max_depth) + + for pkg, deps in tree.items(): + if visited is not None and pkg.key not in visited: + continue + graph.node(pkg.key, label=f"{pkg.project_name}\\n{pkg.version}") + if visited is None or visited[pkg.key] < max_depth: + for dep in deps: + if visited is not None and dep.key not in visited: + continue + if dep.is_missing: + graph.node(dep.key, label=f"{dep.project_name}\\n(missing)", style="dashed") + graph.edge(pkg.key, dep.key, style="dashed") + else: + graph.edge(pkg.key, dep.key, label=dep.edge_label) + + +def _compute_reachable_depths(tree: PackageDAG, root_keys: set[str], max_depth: float) -> dict[str, int] | None: + """BFS from root_keys, returning {key: depth} for all nodes reachable within max_depth.""" + if max_depth >= math.inf: + return None + visited: dict[str, int] = {} + queue: deque[tuple[str, int]] = deque((k, 0) for k in root_keys) + while queue: + key, depth = queue.popleft() + if key in visited: + continue + visited[key] = depth + if depth < max_depth: + for child in tree.get_children(key): + if child.key not in visited: + queue.append((child.key, depth + 1)) + return visited + + +def _get_root_keys(tree: PackageDAG) -> set[str]: + """Return keys that are not dependencies of any other package (i.e. root nodes).""" + dep_keys = {dep.key for deps in tree.values() for dep in deps} + return {pkg.key for pkg in tree if pkg.key not in dep_keys} __all__ = [ diff --git a/pipenv/vendor/pipdeptree/_render/mermaid.py b/pipenv/vendor/pipdeptree/_render/mermaid.py index de88bb520f..1e75b6fb5c 100644 --- a/pipenv/vendor/pipdeptree/_render/mermaid.py +++ b/pipenv/vendor/pipdeptree/_render/mermaid.py @@ -71,32 +71,28 @@ def mermaid_id(key: str) -> str: if isinstance(tree, ReversedPackageDAG): for package, reverse_dependencies in tree.items(): assert isinstance(package, ReqPackage) - package_label = "\\n".join( + package_label = "
".join( (package.project_name, "(missing)" if package.is_missing else package.installed_version), ) package_key = mermaid_id(package.key) nodes.add(f'{package_key}["{package_label}"]') for reverse_dependency in reverse_dependencies: assert isinstance(reverse_dependency, DistPackage) - edge_label = ( - reverse_dependency.req.version_spec if reverse_dependency.req is not None else None - ) or "any" reverse_dependency_key = mermaid_id(reverse_dependency.key) - edges.add(f'{package_key} -- "{edge_label}" --> {reverse_dependency_key}') + edges.add(f'{package_key} -- "{reverse_dependency.edge_label}" --> {reverse_dependency_key}') else: for package, dependencies in tree.items(): - package_label = f"{package.project_name}\\n{package.version}" + package_label = f"{package.project_name}
{package.version}" package_key = mermaid_id(package.key) nodes.add(f'{package_key}["{package_label}"]') for dependency in dependencies: - edge_label = dependency.version_spec or "any" dependency_key = mermaid_id(dependency.key) if dependency.is_missing: - dependency_label = f"{dependency.project_name}\\n(missing)" + dependency_label = f"{dependency.project_name}
(missing)" nodes.add(f'{dependency_key}["{dependency_label}"]:::missing') edges.add(f"{package_key} -.-> {dependency_key}") else: - edges.add(f'{package_key} -- "{edge_label}" --> {dependency_key}') + edges.add(f'{package_key} -- "{dependency.edge_label}" --> {dependency_key}') # Produce the Mermaid Markdown. lines = [ diff --git a/pipenv/vendor/pipdeptree/_render/rich_text.py b/pipenv/vendor/pipdeptree/_render/rich_text.py new file mode 100644 index 0000000000..de0796b1a8 --- /dev/null +++ b/pipenv/vendor/pipdeptree/_render/rich_text.py @@ -0,0 +1,178 @@ +from __future__ import annotations + +import re +import sys +from typing import TYPE_CHECKING + +from pipenv.vendor.pipdeptree._models.package import DistPackage, ReqPackage + +if TYPE_CHECKING: + from rich.tree import Tree + + from pipenv.vendor.pipdeptree._models import PackageDAG + + +def render_rich_text( + tree: PackageDAG, + *, + max_depth: float, + list_all: bool = True, + include_license: bool = False, +) -> None: + """ + Print tree using Rich library for enhanced terminal output. + + :param tree: the package tree + :param max_depth: the maximum depth of the dependency tree + :param list_all: whether to list all the pkgs at the root level or only those that are the sub-dependencies + :param include_license: provide license information + :returns: None + """ + try: + from rich.console import Console # noqa: PLC0415 + from rich.tree import Tree # noqa: PLC0415 + except ImportError as exc: + print( # noqa: T201 + "rich is not available, but necessary for the output option. Please install it.", + file=sys.stderr, + ) + raise SystemExit(1) from exc + + from pipenv.vendor.pipdeptree._render.text import get_top_level_nodes # noqa: PLC0415 + + nodes = get_top_level_nodes(tree, list_all=list_all) + console = Console() + + for node in nodes: + root_label = _format_node(node, parent=None, include_license=include_license) + rich_tree = Tree(root_label, guide_style="bold bright_blue") + _build_tree(tree, node, rich_tree, max_depth=max_depth, depth=0, cur_chain=[]) + console.print(rich_tree) + + +def _build_tree( # noqa: PLR0913 + tree: PackageDAG, + node: DistPackage | ReqPackage, + rich_tree: Tree, + *, + max_depth: float, + depth: int, + cur_chain: list[str], +) -> None: + """ + Recursively build the rich tree structure. + + :param tree: the package tree + :param node: current node + :param rich_tree: the rich Tree object to add children to + :param max_depth: maximum depth + :param depth: current depth + :param cur_chain: chain of package names to detect cycles + """ + if depth >= max_depth: + return + + children = tree.get_children(node.key) + for child in children: + if child.project_name in cur_chain: + continue + + child_label = _format_node(child, parent=node, include_license=False) + child_tree = rich_tree.add(child_label) + + _build_tree( + tree, + child, + child_tree, + max_depth=max_depth, + depth=depth + 1, + cur_chain=[*cur_chain, child.project_name], + ) + + +def _format_node( + node: DistPackage | ReqPackage, + parent: DistPackage | ReqPackage | None, + *, + include_license: bool, +) -> str: + """ + Format a node for display with rich styling. + + :param node: the node to format + :param parent: the parent node (if any) + :param include_license: whether to include license information + :return: formatted string with rich markup + """ + node_str = node.render(parent, frozen=False) + + if parent is None and include_license: + node_str += " " + node.licenses() + + if parent is None: + return _format_root_node(node_str) + return _format_branch_node(node_str, node) + + +def _format_root_node(node_str: str) -> str: + """Format a root node (package at top level).""" + match = re.match(r"^(.+?)==(.+?)(\s+\(.+\))?$", node_str) + assert match, f"Unexpected root node format: {node_str}" + name, version, license_part = match.groups() + license_str = f"[dim]{license_part}[/dim]" if license_part else "" + return f"[bold cyan]{name}[/bold cyan][dim]==[/dim][bold green]{version}[/bold green]{license_str}" + + +def _format_branch_node(node_str: str, node: DistPackage | ReqPackage) -> str: + """Format a branch node (dependency).""" + if isinstance(node, ReqPackage) and ( + match := re.match( + r""" + ^(.+?) # package name (non-greedy) + \s+\[ # opening bracket + required:\s*(.+?) # required version spec (supports multi-spec like >=1.0,<2.0) + ,\s+installed:\s*(.+?) # installed version + (?:,\s+extra:\s*(.+?))? # optional extra name + \]$ # closing bracket + """, + node_str, + re.VERBOSE, + ) + ): + name, required, installed, extra = match.groups() + status_icon = _get_status_icon(node) + extra_str = f" [magenta]\\[extra: {extra}][/magenta]" if extra else "" + return ( + f"{status_icon} [bold cyan]{name}[/bold cyan] " + f"[dim]required:[/dim] [yellow]{required}[/yellow] " + f"[dim]installed:[/dim] {_format_version(installed, node)}{extra_str}" + ) + + match = re.match(r"^(.+?)==(.+?)\s+\[requires:\s*(.+?)\]$", node_str) + assert match, f"Unexpected branch node format: {node_str}" + pkg_name, pkg_version, requires = match.groups() + return ( + f"[bold cyan]{pkg_name}[/bold cyan][dim]==[/dim][bold green]{pkg_version}[/bold green] " + f"[dim]\\[requires:[/dim] [yellow]{requires}[/yellow][dim]][/dim]" + ) + + +def _get_status_icon(node: ReqPackage) -> str: + """Get a status icon for a requirement package.""" + if node.is_missing: + return "[bold red]✗[/bold red]" + if node.is_conflicting(): + return "[bold yellow]⚠[/bold yellow]" + return "[bold green]✓[/bold green]" + + +def _format_version(version: str, node: ReqPackage) -> str: + """Format version with appropriate color based on status.""" + if node.is_missing: + return f"[bold red]{version}[/bold red]" + if node.is_conflicting(): + return f"[bold yellow]{version}[/bold yellow]" + return f"[bold green]{version}[/bold green]" + + +__all__ = ["render_rich_text"] diff --git a/pipenv/vendor/pipdeptree/_render/text.py b/pipenv/vendor/pipdeptree/_render/text.py index 2112b30398..085090120e 100644 --- a/pipenv/vendor/pipdeptree/_render/text.py +++ b/pipenv/vendor/pipdeptree/_render/text.py @@ -31,7 +31,7 @@ def render_text( if encoding in {"utf-8", "utf-16", "utf-32"}: _render_text_with_unicode(tree, nodes, max_depth, include_license) else: - _render_text_without_unicode(tree, nodes, max_depth, include_license) + _render_text_simple(tree, nodes, max_depth, include_license) def get_top_level_nodes(tree: PackageDAG, *, list_all: bool) -> list[DistPackage]: @@ -117,11 +117,14 @@ def aux( # noqa: PLR0913, PLR0917 print("\n".join(lines)) # noqa: T201 -def _render_text_without_unicode( +def _render_text_simple( # noqa: PLR0913 tree: PackageDAG, nodes: list[DistPackage], max_depth: float, include_license: bool, # noqa: FBT001 + *, + frozen: bool = False, + bullet: str = "- ", ) -> None: def aux( node: DistPackage | ReqPackage, @@ -131,10 +134,9 @@ def aux( depth: int = 0, ) -> list[Any]: cur_chain = cur_chain or [] - node_str = node.render(parent, frozen=False) + node_str = node.render(parent, frozen=frozen) if parent: - prefix = " " * indent + "- " - node_str = prefix + node_str + node_str = " " * indent + bullet + node_str elif include_license: node_str += " " + node.licenses() result = [node_str] diff --git a/pipenv/vendor/pipdeptree/_validate.py b/pipenv/vendor/pipdeptree/_validate.py index f06cafa411..0e784abf6d 100644 --- a/pipenv/vendor/pipdeptree/_validate.py +++ b/pipenv/vendor/pipdeptree/_validate.py @@ -4,12 +4,13 @@ from collections import defaultdict from typing import TYPE_CHECKING +from pipenv.vendor.pipdeptree._models.package import ReqPackage from pipenv.vendor.pipdeptree._warning import get_warning_printer if TYPE_CHECKING: from pipenv.vendor.pipdeptree._models.package import Package - from ._models import DistPackage, PackageDAG, ReqPackage + from ._models import DistPackage, PackageDAG def validate(tree: PackageDAG) -> None: @@ -101,16 +102,19 @@ def dfs(root: DistPackage, current: Package, visited: set[str], cdeps: list[Pack def render_cycles_text(cycles: list[list[Package]]) -> None: # List in alphabetical order the dependency that caused the cycle (i.e. the second-to-last Package element) - cycles = sorted(cycles, key=lambda c: c[len(c) - 2].key) + cycles = sorted(cycles, key=lambda c: c[-2].key) for cycle in cycles: print("*", end=" ", file=sys.stderr) # noqa: T201 size = len(cycle) - 1 for idx, pkg in enumerate(cycle): + name = pkg.project_name + if isinstance(pkg, ReqPackage) and pkg.extra: + name += f" [extra: {pkg.extra}]" if idx == size: - print(f"{pkg.project_name}", end="", file=sys.stderr) # noqa: T201 + print(name, end="", file=sys.stderr) # noqa: T201 else: - print(f"{pkg.project_name} =>", end=" ", file=sys.stderr) # noqa: T201 + print(f"{name} =>", end=" ", file=sys.stderr) # noqa: T201 print(file=sys.stderr) # noqa: T201 diff --git a/pipenv/vendor/pipdeptree/_warning.py b/pipenv/vendor/pipdeptree/_warning.py index 0ce5810b96..911dd3586c 100644 --- a/pipenv/vendor/pipdeptree/_warning.py +++ b/pipenv/vendor/pipdeptree/_warning.py @@ -15,14 +15,11 @@ class WarningType(Enum): @classmethod def from_str(cls, string: str) -> WarningType: - if string == "silence": - return WarningType.SILENCE - if string == "suppress": - return WarningType.SUPPRESS - if string == "fail": - return WarningType.FAIL - msg = "Unknown WarningType string value provided" - raise ValueError(msg) + try: + return cls(string) + except ValueError: + msg = "Unknown WarningType string value provided" + raise ValueError(msg) from None class WarningPrinter: diff --git a/pipenv/vendor/pipdeptree/version.py b/pipenv/vendor/pipdeptree/version.py index af464aa055..1bbb20f93f 100644 --- a/pipenv/vendor/pipdeptree/version.py +++ b/pipenv/vendor/pipdeptree/version.py @@ -28,7 +28,7 @@ commit_id: COMMIT_ID __commit_id__: COMMIT_ID -__version__ = version = '2.31.0' -__version_tuple__ = version_tuple = (2, 31, 0) +__version__ = version = '2.34.0' +__version_tuple__ = version_tuple = (2, 34, 0) __commit_id__ = commit_id = None diff --git a/pipenv/vendor/pythonfinder/__init__.py b/pipenv/vendor/pythonfinder/__init__.py index dd2b6688b8..b1f028ebb1 100644 --- a/pipenv/vendor/pythonfinder/__init__.py +++ b/pipenv/vendor/pythonfinder/__init__.py @@ -4,6 +4,6 @@ from .models.python_info import PythonInfo from .pythonfinder import Finder -__version__ = "3.0.2" +__version__ = "3.0.3" __all__ = ["Finder", "PythonInfo", "InvalidPythonVersion", "PythonNotFound"] diff --git a/pipenv/vendor/pythonfinder/finders/py_launcher_finder.py b/pipenv/vendor/pythonfinder/finders/py_launcher_finder.py index ce08fafdbe..ae4afe5e1c 100644 --- a/pipenv/vendor/pythonfinder/finders/py_launcher_finder.py +++ b/pipenv/vendor/pythonfinder/finders/py_launcher_finder.py @@ -92,7 +92,10 @@ def _create_python_info_from_py_launcher( Create a PythonInfo object from py launcher information. Args: - version: The Python version (e.g. "3.11"). + version: The Python version reported by the py launcher (e.g. "3.11"). + This is typically only major.minor; the full patch version is + resolved by querying the executable so that searches for a + specific full version (e.g. "3.11.9") succeed. path: The path to the Python executable. is_default: "*" if this is the default version, "" otherwise. @@ -102,7 +105,7 @@ def _create_python_info_from_py_launcher( if not path or not os.path.exists(path): return None - # Parse the version + # Parse the version reported by the py launcher (usually major.minor only). try: version_data = parse_python_version(version) except InvalidPythonVersion: @@ -110,6 +113,22 @@ def _create_python_info_from_py_launcher( raise return None + # `py --list-paths` only reports major.minor (e.g. "3.11"). Resolve the + # real patch version by querying the executable so that callers searching + # for a full version string like "3.11.9" can match correctly. + if version_data.get("patch") is None: + try: + from ..utils.version_utils import get_python_version + + full_version = get_python_version(path) + full_version_data = parse_python_version(full_version) + version_data = full_version_data + version = full_version + except Exception: + # If we can't run the executable, fall back to the major.minor + # version string provided by the py launcher. + pass + # Create the PythonInfo object return PythonInfo( path=Path(path), diff --git a/pipenv/vendor/vendor.txt b/pipenv/vendor/vendor.txt index d388c1bace..de5be43195 100644 --- a/pipenv/vendor/vendor.txt +++ b/pipenv/vendor/vendor.txt @@ -1,14 +1,14 @@ click-didyoumean==0.3.1 click==8.3.1 colorama==0.4.6 -importlib-metadata==8.7.1 +importlib-metadata==9.0.0 zipp==3.23.0 packaging==26.0 pexpect==4.9.0 -pipdeptree==2.31.0 +pipdeptree==2.34.0 plette==2.1.0 ptyprocess==0.7.0 python-dotenv==1.2.2 -pythonfinder==3.0.2 +pythonfinder==3.0.3 shellingham==1.5.4 tomlkit==0.14.0