Bump vendored pipdeptree

Also update the patch we apply

Signed-off-by: Oz Tiram <oz.tiram@gmail.com>
This commit is contained in:
Oz Tiram
2024-01-04 14:56:56 +01:00
parent e45cc7a9ab
commit d5564b082a
19 changed files with 1421 additions and 1091 deletions
File diff suppressed because it is too large Load Diff
+45 -1
View File
@@ -1,5 +1,17 @@
"""The main entry point used for CLI."""
from __future__ import annotations
import os
import sys
from typing import Sequence
from pipenv.vendor.pipdeptree._cli import get_options
from pipenv.vendor.pipdeptree._discovery import get_installed_distributions
from pipenv.vendor.pipdeptree._models import PackageDAG
from pipenv.vendor.pipdeptree._non_host import handle_non_host_target
from pipenv.vendor.pipdeptree._render import render
from pipenv.vendor.pipdeptree._validate import validate
pardir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# for finding pipdeptree itself
@@ -8,7 +20,39 @@ sys.path.append(pardir)
sys.path.append(os.path.dirname(os.path.dirname(pardir)))
from pipenv.vendor.pipdeptree import main
def main(args: Sequence[str] | None = None) -> None | int:
"""CLI - The main function called as entry point."""
options = get_options(args)
result = handle_non_host_target(options)
if result is not None:
return result
pkgs = get_installed_distributions(local_only=options.local_only, user_only=options.user_only)
tree = PackageDAG.from_pkgs(pkgs)
is_text_output = not any([options.json, options.json_tree, options.output_format])
return_code = validate(options, is_text_output, tree)
# Reverse the tree (if applicable) before filtering, thus ensuring, that the filter will be applied on ReverseTree
if options.reverse:
tree = tree.reverse()
show_only = options.packages.split(",") if options.packages else None
exclude = set(options.exclude.split(",")) if options.exclude else None
if show_only is not None or exclude is not None:
try:
tree = tree.filter_nodes(show_only, exclude)
except ValueError as e:
if options.warn in ("suppress", "fail"):
print(e, file=sys.stderr) # noqa: T201
return_code |= 1 if options.warn == "fail" else 0
return return_code
render(options, tree)
return return_code
if __name__ == "__main__":
sys.exit(main())
+152
View File
@@ -0,0 +1,152 @@
from __future__ import annotations
import sys
from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser, Namespace
from typing import TYPE_CHECKING, Sequence, cast
from .version import __version__
if TYPE_CHECKING:
from typing import Literal
class Options(Namespace):
freeze: bool
python: str
all: bool # noqa: A003
local_only: bool
user_only: bool
warn: Literal["silence", "suppress", "fail"]
reverse: bool
packages: str
exclude: str
json: bool
json_tree: bool
mermaid: bool
output_format: str | None
depth: float
encoding: str
class _Formatter(ArgumentDefaultsHelpFormatter):
def __init__(self, prog: str) -> None:
super().__init__(prog, max_help_position=22, width=240)
def build_parser() -> ArgumentParser:
parser = ArgumentParser(description="Dependency tree of the installed python packages", formatter_class=_Formatter)
parser.add_argument("-v", "--version", action="version", version=f"{__version__}")
parser.add_argument(
"-w",
"--warn",
action="store",
dest="warn",
nargs="?",
default="suppress",
choices=("silence", "suppress", "fail"),
help=(
"warning control: suppress will show warnings but return 0 whether or not they are present; silence will "
"not show warnings at all and always return 0; fail will show warnings and return 1 if any are present"
),
)
parser.add_argument(
"-r",
"--reverse",
action="store_true",
default=False,
help=(
"render the dependency tree in the reverse fashion ie. the sub-dependencies are listed with the list of "
"packages that need them under them"
),
)
select = parser.add_argument_group(title="select", description="choose what to render")
select.add_argument("--python", default=sys.executable, help="Python interpreter to inspect")
select.add_argument(
"-p",
"--packages",
help="comma separated list of packages to show - wildcards are supported, like 'somepackage.*'",
metavar="P",
)
select.add_argument(
"-e",
"--exclude",
help="comma separated list of packages to not show - wildcards are supported, like 'somepackage.*'. "
"(cannot combine with -p or -a)",
metavar="P",
)
select.add_argument("-a", "--all", action="store_true", help="list all deps at top level")
scope = select.add_mutually_exclusive_group()
scope.add_argument(
"-l",
"--local-only",
action="store_true",
help="if in a virtualenv that has global access do not show globally installed packages",
)
scope.add_argument("-u", "--user-only", action="store_true", help="only show installations in the user site dir")
render = parser.add_argument_group(
title="render",
description="choose how to render the dependency tree (by default will use text mode)",
)
render.add_argument("-f", "--freeze", action="store_true", help="print names so as to write freeze files")
render.add_argument(
"--encoding",
dest="encoding_type",
default=sys.stdout.encoding,
help="the encoding to use when writing to the output",
metavar="E",
)
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 render only)",
metavar="D",
)
render_type = render.add_mutually_exclusive_group()
render_type.add_argument(
"-j",
"--json",
action="store_true",
default=False,
help="raw JSON - this will yield output that may be used by external tools",
)
render_type.add_argument(
"--json-tree",
action="store_true",
default=False,
help="nested JSON - mimics the text format layout",
)
render_type.add_argument(
"--mermaid",
action="store_true",
default=False,
help="https://mermaid.js.org flow diagram",
)
render_type.add_argument(
"--graph-output",
metavar="FMT",
dest="output_format",
help="Graphviz rendering with the value being the graphviz output e.g.: dot, jpeg, pdf, png, svg",
)
return parser
def get_options(args: Sequence[str] | None) -> Options:
parser = build_parser()
parsed_args = parser.parse_args(args)
if parsed_args.exclude and (parsed_args.all or parsed_args.packages):
return parser.error("cannot use --exclude with --packages or --all")
return cast(Options, parsed_args)
__all__ = [
"get_options",
"Options",
]
+37
View File
@@ -0,0 +1,37 @@
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from pipenv.patched.pip._vendor.pkg_resources import DistInfoDistribution
def get_installed_distributions(
local_only: bool = False, # noqa: FBT001, FBT002
user_only: bool = False, # noqa: FBT001, FBT002
) -> list[DistInfoDistribution]:
try:
from pipenv.patched.pip._internal.metadata import pkg_resources
except ImportError:
# For backward compatibility with python ver. 2.7 and pip
# version 20.3.4 (the latest pip version that works with python
# version 2.7)
from pipenv.patched.pip._internal.utils import misc
return misc.get_installed_distributions( # type: ignore[no-any-return,attr-defined]
local_only=local_only,
user_only=user_only,
)
else:
dists = pkg_resources.Environment.from_paths(None).iter_installed_distributions(
local_only=local_only,
skip=(),
user_only=user_only,
)
return [d._dist for d in dists] # type: ignore[attr-defined] # noqa: SLF001
__all__ = [
"get_installed_distributions",
]
+11
View File
@@ -0,0 +1,11 @@
from __future__ import annotations
from .dag import PackageDAG, ReversedPackageDAG
from .package import DistPackage, ReqPackage
__all__ = [
"ReqPackage",
"DistPackage",
"PackageDAG",
"ReversedPackageDAG",
]
+246
View File
@@ -0,0 +1,246 @@
from __future__ import annotations
from collections import defaultdict, deque
from fnmatch import fnmatch
from itertools import chain
from typing import TYPE_CHECKING, Iterator, List, Mapping
if TYPE_CHECKING:
from pipenv.patched.pip._vendor.pkg_resources import DistInfoDistribution
from .package import DistPackage, ReqPackage
class PackageDAG(Mapping[DistPackage, List[ReqPackage]]):
"""Representation of Package dependencies as directed acyclic graph using a dict as the underlying datastructure.
The nodes and their relationships (edges) are internally stored using a map as follows,
{a: [b, c],
b: [d],
c: [d, e],
d: [e],
e: [],
f: [b],
g: [e, f]}
Here, node `a` has 2 children nodes `b` and `c`. Consider edge direction from `a` -> `b` and `a` -> `c`
respectively.
A node is expected to be an instance of a subclass of `Package`. The keys are must be of class `DistPackage` and
each item in values must be of class `ReqPackage`. (See also ReversedPackageDAG where the key and value types are
interchanged).
"""
@classmethod
def from_pkgs(cls, pkgs: list[DistInfoDistribution]) -> PackageDAG:
dist_pkgs = [DistPackage(p) for p in pkgs]
idx = {p.key: p for p in dist_pkgs}
m: dict[DistPackage, list[ReqPackage]] = {}
for p in dist_pkgs:
reqs = []
for r in p.requires():
d = idx.get(r.key)
# pip's _vendor.packaging.requirements.Requirement uses the exact casing of a dependency's name found in
# a project's build config, which is not ideal when rendering.
# See https://github.com/tox-dev/pipdeptree/issues/242
r.project_name = d.project_name if d is not None else r.project_name
pkg = ReqPackage(r, d)
reqs.append(pkg)
m[p] = reqs
return cls(m)
def __init__(self, m: dict[DistPackage, list[ReqPackage]]) -> None:
"""Initialize the PackageDAG object.
:param dict m: dict of node objects (refer class docstring)
:returns: None
:rtype: NoneType
"""
self._obj: dict[DistPackage, list[ReqPackage]] = m
self._index: dict[str, DistPackage] = {p.key: p for p in list(self._obj)}
def get_node_as_parent(self, node_key: str) -> DistPackage | None:
"""Get the node from the keys of the dict representing the DAG.
This method is useful if the dict representing the DAG contains different kind of objects in keys and values.
Use this method to look up a node obj as a parent (from the keys of the dict) given a node key.
:param node_key: identifier corresponding to key attr of node obj
:returns: node obj (as present in the keys of the dict)
"""
try:
return self._index[node_key]
except KeyError:
return None
def get_children(self, node_key: str) -> list[ReqPackage]:
"""Get child nodes for a node by its key.
:param node_key: key of the node to get children of
:returns: child nodes
"""
node = self.get_node_as_parent(node_key)
return self._obj[node] if node else []
def filter_nodes(self, include: list[str] | None, exclude: set[str] | None) -> PackageDAG: # noqa: C901, PLR0912
"""Filter nodes in a graph by given parameters.
If a node is included, then all it's children are also included.
:param include: list of node keys to include (or None)
:param exclude: set of node keys to exclude (or None)
:raises ValueError: If include has node keys that do not exist in the graph
:returns: filtered version of the graph
"""
# If neither of the filters are specified, short circuit
if include is None and exclude is None:
return self
# Note: In following comparisons, we use lower cased values so
# that user may specify `key` or `project_name`. As per the
# documentation, `key` is simply
# `project_name.lower()`. Refer:
# https://setuptools.readthedocs.io/en/latest/pkg_resources.html#distribution-objects
include_with_casing_preserved: list[str] = []
if include:
include_with_casing_preserved = include
include = [s.lower() for s in include]
exclude = {s.lower() for s in exclude} if exclude else set()
# Check for mutual exclusion of show_only and exclude sets
# after normalizing the values to lowercase
if include and exclude:
assert not (set(include) & exclude)
# Traverse the graph in a depth first manner and filter the
# nodes according to `show_only` and `exclude` sets
stack: deque[DistPackage] = deque()
m: dict[DistPackage, list[ReqPackage]] = {}
seen = set()
matched_includes: set[str] = set()
for node in self._obj:
if any(fnmatch(node.key, e) for e in exclude):
continue
if include is None:
stack.append(node)
else:
should_append = False
for i in include:
if fnmatch(node.key, i):
# Add all patterns that match with the node key. Otherwise if we break, patterns like py* or
# pytest* (which both should match "pytest") may cause one pattern to be missed and will
# raise an error
matched_includes.add(i)
should_append = True
if should_append:
stack.append(node)
while stack:
n = stack.pop()
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:
if c.key not in seen:
cld_node = self.get_node_as_parent(c.key)
if cld_node:
stack.append(cld_node)
else:
# It means there's no root node corresponding to the child node i.e.
# a dependency is missing
continue
non_existent_includes = [i for i in include_with_casing_preserved if i.lower() not in matched_includes]
if non_existent_includes:
raise ValueError("No packages matched using the following patterns: " + ", ".join(non_existent_includes))
return self.__class__(m)
def reverse(self) -> ReversedPackageDAG:
"""Reverse the DAG, or turn it upside-down.
In other words, the directions of edges of the nodes in the DAG will be reversed.
Note that this function purely works on the nodes in the graph. This implies that to perform a combination of
filtering and reversing, the order in which `filter` and `reverse` methods should be applied is important. For
e.g., if reverse is called on a filtered graph, then only the filtered nodes and it's children will be
considered when reversing. On the other hand, if filter is called on reversed DAG, then the definition of
"child" nodes is as per the reversed DAG.
:returns: DAG in the reversed form
"""
m: defaultdict[ReqPackage, list[DistPackage]] = defaultdict(list)
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]
def sort(self) -> PackageDAG:
"""Return sorted tree in which the underlying _obj dict is an dict, sorted alphabetically by the keys.
:returns: Instance of same class with dict
"""
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 __iter__(self) -> Iterator[DistPackage]:
return self._obj.__iter__()
def __len__(self) -> int:
return len(self._obj)
class ReversedPackageDAG(PackageDAG):
"""Representation of Package dependencies in the reverse order.
Similar to it's super class `PackageDAG`, the underlying datastructure is a dict, but here the keys are expected to
be of type `ReqPackage` and each item in the values of type `DistPackage`.
Typically, this object will be obtained by calling `PackageDAG.reverse`.
"""
def reverse(self) -> PackageDAG: # type: ignore[override]
"""Reverse the already reversed DAG to get the PackageDAG again.
:returns: reverse of the reversed DAG
"""
m: defaultdict[DistPackage, list[ReqPackage]] = defaultdict(list)
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) # type: ignore[arg-type]
if k.key not in child_keys:
assert isinstance(k, ReqPackage)
assert k.dist is not None
m[k.dist] = []
return PackageDAG(dict(m))
__all__ = [
"PackageDAG",
"ReversedPackageDAG",
]
+228
View File
@@ -0,0 +1,228 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from importlib import import_module
from importlib.metadata import PackageNotFoundError, version
from inspect import ismodule
from typing import TYPE_CHECKING
from pipenv.patched.pip._vendor.pkg_resources import Requirement
if TYPE_CHECKING:
from pipenv.patched.pip._internal.metadata import BaseDistribution
from pipenv.patched.pip._vendor.pkg_resources import DistInfoDistribution
class Package(ABC):
"""Abstract class for wrappers around objects that pip returns."""
def __init__(self, obj: DistInfoDistribution) -> None:
self._obj: DistInfoDistribution = obj
@property
def key(self) -> str:
return self._obj.key # type: ignore[no-any-return]
@property
def project_name(self) -> str:
return self._obj.project_name # type: ignore[no-any-return]
@abstractmethod
def render_as_root(self, *, frozen: bool) -> str:
raise NotImplementedError
@abstractmethod
def render_as_branch(self, *, frozen: bool) -> str:
raise NotImplementedError
@abstractmethod
def as_dict(self) -> dict[str, str | None]:
raise NotImplementedError
@property
def version_spec(self) -> None | str:
return None
def render(
self,
parent: DistPackage | ReqPackage | None = None,
*,
frozen: bool = False,
) -> str:
render = self.render_as_branch if parent else self.render_as_root
return render(frozen=frozen)
@staticmethod
def as_frozen_repr(obj: DistInfoDistribution) -> str:
# The `pipenv.patched.pip._internal.metadata` modules were introduced in 21.1.1
# and the `pipenv.patched.pip._internal.operations.freeze.FrozenRequirement`
# class now expects dist to be a subclass of
# `pipenv.patched.pip._internal.metadata.BaseDistribution`, however the
# `pipenv.patched.pip._internal.utils.misc.get_installed_distributions` continues
# to return objects of type
# pipenv.patched.pip._vendor.pkg_resources.DistInfoDistribution.
#
# This is a hacky backward compatible (with older versions of pip) fix.
try:
from pipenv.patched.pip._internal.operations.freeze import FrozenRequirement
except ImportError:
from pipenv.patched.pip import FrozenRequirement # type: ignore[attr-defined, no-redef]
try:
from pipenv.patched.pip._internal import metadata
except ImportError:
our_dist: BaseDistribution = obj # type: ignore[assignment]
else:
our_dist = metadata.pkg_resources.Distribution(obj)
try:
fr = FrozenRequirement.from_dist(our_dist)
except TypeError:
fr = FrozenRequirement.from_dist(our_dist, []) # type: ignore[call-arg]
return str(fr).strip()
def __repr__(self) -> str:
return f'<{self.__class__.__name__}("{self.key}")>'
def __lt__(self, rhs: Package) -> bool:
return self.key < rhs.key
class DistPackage(Package):
"""Wrapper class for pkg_resources.Distribution instances.
:param obj: pkg_resources.Distribution to wrap over
:param req: optional ReqPackage object to associate this DistPackage with. This is useful for displaying the tree in
reverse
"""
def __init__(self, obj: DistInfoDistribution, req: ReqPackage | None = None) -> None:
super().__init__(obj)
self.req = req
def requires(self) -> list[Requirement]:
return self._obj.requires() # type: ignore[no-untyped-call,no-any-return]
@property
def version(self) -> str:
return self._obj.version # type: ignore[no-any-return]
def render_as_root(self, *, frozen: bool) -> str:
if not frozen:
return f"{self.project_name}=={self.version}"
return self.as_frozen_repr(self._obj)
def render_as_branch(self, *, frozen: bool) -> str:
assert self.req is not None
if not frozen:
parent_ver_spec = self.req.version_spec
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}]"
return self.render_as_root(frozen=frozen)
def as_requirement(self) -> ReqPackage:
"""Return a ReqPackage representation of this DistPackage."""
return ReqPackage(self._obj.as_requirement(), dist=self) # type: ignore[no-untyped-call]
def as_parent_of(self, req: ReqPackage | None) -> DistPackage:
"""Return a DistPackage instance associated to a requirement.
This association is necessary for reversing the PackageDAG.
If `req` is None, and the `req` attribute of the current instance is also None, then the same instance will be
returned.
:param ReqPackage req: the requirement to associate with
:returns: DistPackage instance
"""
if req is None and self.req is None:
return self
return self.__class__(self._obj, req)
def as_dict(self) -> dict[str, str | None]:
return {"key": self.key, "package_name": self.project_name, "installed_version": self.version}
class ReqPackage(Package):
"""Wrapper class for Requirements instance.
:param obj: The `Requirements` instance to wrap over
:param dist: optional `pkg_resources.Distribution` instance for this requirement
"""
UNKNOWN_VERSION = "?"
def __init__(self, obj: Requirement, dist: DistPackage | None = None) -> None:
super().__init__(obj)
self.dist = dist
def render_as_root(self, *, frozen: bool) -> str:
if not frozen:
return f"{self.project_name}=={self.installed_version}"
if self.dist:
return self.as_frozen_repr(self.dist._obj) # noqa: SLF001
return self.project_name
def render_as_branch(self, *, frozen: bool) -> str:
if not frozen:
req_ver = self.version_spec if self.version_spec else "Any"
return f"{self.project_name} [required: {req_ver}, installed: {self.installed_version}]"
return self.render_as_root(frozen=frozen)
@property
def version_spec(self) -> str | None:
specs = sorted(self._obj.specs, reverse=True) # `reverse` makes '>' prior to '<'
return ",".join(["".join(sp) for sp in specs]) if specs else None
@property
def installed_version(self) -> str:
if not self.dist:
try:
return version(self.key)
except PackageNotFoundError:
pass
# Avoid AssertionError with setuptools, see https://github.com/tox-dev/pipdeptree/issues/162
if self.key in {"setuptools"}:
return self.UNKNOWN_VERSION
try:
m = import_module(self.key)
except ImportError:
return self.UNKNOWN_VERSION
else:
v = getattr(m, "__version__", self.UNKNOWN_VERSION)
if ismodule(v):
return getattr(v, "__version__", self.UNKNOWN_VERSION)
return v
return self.dist.version
@property
def is_missing(self) -> bool:
return self.installed_version == self.UNKNOWN_VERSION
def is_conflicting(self) -> bool:
"""If installed version conflicts with required version."""
# unknown installed version is also considered conflicting
if self.installed_version == self.UNKNOWN_VERSION:
return True
ver_spec = self.version_spec if self.version_spec else ""
req_version_str = f"{self.project_name}{ver_spec}"
req_obj = Requirement.parse(req_version_str) # type: ignore[no-untyped-call]
return self.installed_version not in req_obj
def as_dict(self) -> dict[str, str | None]:
return {
"key": self.key,
"package_name": self.project_name,
"installed_version": self.installed_version,
"required_version": self.version_spec,
}
__all__ = [
"DistPackage",
"ReqPackage",
]
+52
View File
@@ -0,0 +1,52 @@
from __future__ import annotations
import os
import sys
from inspect import getsourcefile
from pathlib import Path
from shutil import copytree
from subprocess import call
from tempfile import TemporaryDirectory
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ._cli import Options
def handle_non_host_target(args: Options) -> int | None:
# if target is not current python re-invoke it under the actual host
py_path = Path(args.python).absolute()
if py_path != Path(sys.executable).absolute():
# there's no way to guarantee that graphviz is available, so refuse
if args.output_format:
print( # noqa: T201
"graphviz functionality is not supported when querying non-host python",
file=sys.stderr,
)
raise SystemExit(1)
argv = sys.argv[1:] # remove current python executable
for py_at, value in enumerate(argv):
if value == "--python":
del argv[py_at]
del argv[py_at]
elif value.startswith("--python"):
del argv[py_at]
src = getsourcefile(sys.modules[__name__])
assert src is not None
our_root = Path(src).parent
with TemporaryDirectory() as project:
dest = Path(project)
copytree(our_root, dest / "pipdeptree")
# invoke from an empty folder to avoid cwd altering sys.path
env = os.environ.copy()
env["PYTHONPATH"] = project
cmd = [str(py_path), "-m", "pipdeptree", *argv]
return call(cmd, cwd=project, env=env) # noqa: S603
return None
__all__ = [
"handle_non_host_target",
]
+38
View File
@@ -0,0 +1,38 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from .graphviz import render_graphviz
from .json import render_json
from .json_tree import render_json_tree
from .mermaid import render_mermaid
from .text import render_text
if TYPE_CHECKING:
from pipenv.vendor.pipdeptree._cli import Options
from pipenv.vendor.pipdeptree._models import PackageDAG
def render(options: Options, tree: PackageDAG) -> None:
if options.json:
print(render_json(tree)) # noqa: T201
elif options.json_tree:
print(render_json_tree(tree)) # noqa: T201
elif options.mermaid:
print(render_mermaid(tree)) # noqa: T201
elif options.output_format:
assert options.output_format is not None
render_graphviz(tree, output_format=options.output_format, reverse=options.reverse)
else:
render_text(
tree,
max_depth=options.depth,
encoding=options.encoding_type,
list_all=options.all,
frozen=options.freeze,
)
__all__ = [
"render",
]
+116
View File
@@ -0,0 +1,116 @@
from __future__ import annotations
import os
import sys
from typing import TYPE_CHECKING
from pipenv.vendor.pipdeptree._models import DistPackage, ReqPackage
if TYPE_CHECKING:
from pipenv.vendor.pipdeptree._models import PackageDAG
def dump_graphviz( # noqa: C901, PLR0912
tree: PackageDAG,
output_format: str = "dot",
is_reverse: bool = False, # noqa: FBT001, FBT002
) -> str | bytes:
"""Output dependency graph as one of the supported GraphViz output formats.
:param dict tree: dependency graph
:param string output_format: output format
:param bool is_reverse: reverse or not
:returns: representation of tree in the specified output format
:rtype: str or binary representation depending on the output format
"""
try:
from graphviz import Digraph
except ImportError as exc:
print( # noqa: T201
"graphviz is not available, but necessary for the output option. Please install it.",
file=sys.stderr,
)
raise SystemExit(1) from exc
try:
from graphviz import parameters
except ImportError:
from graphviz import backend
valid_formats = backend.FORMATS
print( # noqa: T201
"Deprecation warning! Please upgrade graphviz to version >=0.18.0 "
"Support for older versions will be removed in upcoming release",
file=sys.stderr,
)
else:
valid_formats = parameters.FORMATS
if output_format not in valid_formats:
print(f"{output_format} is not a supported output format.", file=sys.stderr) # noqa: T201
print(f"Supported formats are: {', '.join(sorted(valid_formats))}", file=sys.stderr) # noqa: T201
raise SystemExit(1)
graph = Digraph(format=output_format)
if is_reverse:
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:
# req reference of the dep associated with this particular parent package
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)
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)
# Allow output of dot format, even if GraphViz isn't installed.
if output_format == "dot":
# Emulates graphviz.dot.Dot.__iter__() to force the sorting of graph.body.
# Fixes https://github.com/tox-dev/pipdeptree/issues/188
# That way we can guarantee the output of the dot format is deterministic
# and stable.
return "".join([next(iter(graph)), *sorted(graph.body), graph._tail]) # noqa: SLF001
# 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]
except UnicodeDecodeError:
return graph.pipe() # type: ignore[no-any-return]
def print_graphviz(dump_output: str | bytes) -> None:
"""Dump the data generated by GraphViz to stdout.
:param dump_output: The output from dump_graphviz
"""
if hasattr(dump_output, "encode"):
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) -> None:
output = dump_graphviz(tree, output_format=output_format, is_reverse=reverse)
print_graphviz(output)
__all__ = [
"render_graphviz",
]
+30
View File
@@ -0,0 +1,30 @@
from __future__ import annotations
import json
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from pipenv.vendor.pipdeptree._models import PackageDAG
def render_json(tree: PackageDAG) -> str:
"""Convert the tree into a flat json representation.
The json repr will be a list of hashes, each hash having 2 fields:
- package
- dependencies: list of dependencies
:param tree: dependency tree
:returns: JSON representation of the tree
"""
tree = tree.sort()
return json.dumps(
[{"package": k.as_dict(), "dependencies": [v.as_dict() for v in vs]} for k, vs in tree.items()],
indent=4,
)
__all__ = [
"render_json",
]
+57
View File
@@ -0,0 +1,57 @@
from __future__ import annotations
import json
from itertools import chain
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from pipenv.vendor.pipdeptree._models import DistPackage, PackageDAG, ReqPackage
def render_json_tree(tree: PackageDAG) -> str:
"""Convert the tree into a nested json representation.
The json repr will be a list of hashes, each hash having the following fields:
- package_name
- key
- required_version
- installed_version
- dependencies: list of dependencies
:param tree: dependency tree
:returns: json representation of the tree
"""
tree = tree.sort()
branch_keys = {r.key for r in chain.from_iterable(tree.values())}
nodes = [p for p in tree if p.key not in branch_keys]
def aux(
node: DistPackage | ReqPackage,
parent: DistPackage | ReqPackage | None = None,
cur_chain: list[str] | None = None,
) -> dict[str, Any]:
if cur_chain is None:
cur_chain = [node.project_name]
d: dict[str, str | list[Any] | None] = node.as_dict() # type: ignore[assignment]
if parent:
d["required_version"] = node.version_spec if node.version_spec else "Any"
else:
d["required_version"] = d["installed_version"]
d["dependencies"] = [
aux(c, parent=node, cur_chain=[*cur_chain, c.project_name])
for c in tree.get_children(node.key)
if c.project_name not in cur_chain
]
return d
return json.dumps([aux(p) for p in nodes], indent=4)
__all__ = [
"render_json_tree",
]
+112
View File
@@ -0,0 +1,112 @@
from __future__ import annotations
import itertools as it
from typing import TYPE_CHECKING, Final
from pipenv.vendor.pipdeptree._models import DistPackage, ReqPackage, ReversedPackageDAG
if TYPE_CHECKING:
from pipenv.vendor.pipdeptree._models import PackageDAG
_RESERVED_IDS: Final[frozenset[str]] = frozenset(
[
"C4Component",
"C4Container",
"C4Deployment",
"C4Dynamic",
"_blank",
"_parent",
"_self",
"_top",
"call",
"class",
"classDef",
"click",
"end",
"flowchart",
"flowchart-v2",
"graph",
"interpolate",
"linkStyle",
"style",
"subgraph",
],
)
def render_mermaid(tree: PackageDAG) -> str: # noqa: C901
"""Produce a Mermaid flowchart from the dependency graph.
:param tree: dependency graph
"""
# List of reserved keywords in Mermaid that cannot be used as node names.
# See: https://github.com/mermaid-js/mermaid/issues/4182#issuecomment-1454787806
node_ids_map: dict[str, str] = {}
def mermaid_id(key: str) -> str:
"""Return a valid Mermaid node ID from a string."""
# If we have already seen this key, return the canonical ID.
canonical_id = node_ids_map.get(key)
if canonical_id is not None:
return canonical_id
# If the key is not a reserved keyword, return it as is, and update the map.
if key not in _RESERVED_IDS:
node_ids_map[key] = key
return key
# If the key is a reserved keyword, append a number to it.
for number in it.count():
new_id = f"{key}_{number}"
if new_id not in node_ids_map:
node_ids_map[key] = new_id
return new_id
raise NotImplementedError
# Use a sets to avoid duplicate entries.
nodes: set[str] = set()
edges: set[str] = set()
if isinstance(tree, ReversedPackageDAG):
for package, reverse_dependencies in tree.items():
assert isinstance(package, ReqPackage)
package_label = "\\n".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}')
else:
for package, dependencies in tree.items():
package_label = f"{package.project_name}\\n{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)"
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}')
# Produce the Mermaid Markdown.
lines = [
"flowchart TD",
"classDef missing stroke-dasharray: 5",
*sorted(nodes),
*sorted(edges),
]
return "".join(f"{' ' if i else ''}{line}\n" for i, line in enumerate(lines))
__all__ = [
"render_mermaid",
]
+146
View File
@@ -0,0 +1,146 @@
from __future__ import annotations
from itertools import chain
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from pipenv.vendor.pipdeptree._models import DistPackage, PackageDAG, ReqPackage
def render_text(
tree: PackageDAG,
*,
max_depth: float,
encoding: str,
list_all: bool = True,
frozen: bool = False,
) -> None:
"""Print tree as text on console.
:param tree: the package tree
:param list_all: whether to list all the pgks at the root level or only those that are the sub-dependencies
:param frozen: show the names of the pkgs in the output that's favourable to pip --freeze
:returns: None
"""
tree = tree.sort()
nodes = list(tree.keys())
branch_keys = {r.key for r in chain.from_iterable(tree.values())}
if not list_all:
nodes = [p for p in nodes if p.key not in branch_keys]
if encoding in ("utf-8", "utf-16", "utf-32"):
_render_text_with_unicode(tree, nodes, max_depth, frozen)
else:
_render_text_without_unicode(tree, nodes, max_depth, frozen)
def _render_text_with_unicode(
tree: PackageDAG,
nodes: list[DistPackage],
max_depth: float,
frozen: bool, # noqa: FBT001
) -> None:
use_bullets = not frozen
def aux( # noqa: PLR0913
node: DistPackage | ReqPackage,
parent: DistPackage | ReqPackage | None = None,
indent: int = 0,
cur_chain: list[str] | None = None,
prefix: str = "",
depth: int = 0,
has_grand_parent: bool = False, # noqa: FBT001, FBT002
is_last_child: bool = False, # noqa: FBT001, FBT002
parent_is_last_child: bool = False, # noqa: FBT001, FBT002
) -> list[Any]:
cur_chain = cur_chain or []
node_str = node.render(parent, frozen=frozen)
next_prefix = ""
next_indent = indent + 2
if parent:
bullet = "├── "
if is_last_child:
bullet = "└── "
line_char = ""
if not use_bullets:
line_char = ""
# Add 2 spaces so direct dependencies to a project are indented
bullet = " "
if has_grand_parent:
next_indent -= 1
if parent_is_last_child:
offset = 0 if len(line_char) == 1 else 1
prefix += " " * (indent + 1 - offset - depth)
else:
prefix += line_char + " " * (indent - depth)
# Without this extra space, bullets will point to the space just before the project name
prefix += " " if use_bullets else ""
next_prefix = prefix
node_str = prefix + bullet + node_str
result = [node_str]
children = tree.get_children(node.key)
children_strings = [
aux(
c,
node,
indent=next_indent,
cur_chain=[*cur_chain, c.project_name],
prefix=next_prefix,
depth=depth + 1,
has_grand_parent=parent is not None,
is_last_child=c is children[-1],
parent_is_last_child=is_last_child,
)
for c in children
if c.project_name not in cur_chain and depth + 1 <= max_depth
]
result += list(chain.from_iterable(children_strings))
return result
lines = chain.from_iterable([aux(p) for p in nodes])
print("\n".join(lines)) # noqa: T201
def _render_text_without_unicode(
tree: PackageDAG,
nodes: list[DistPackage],
max_depth: float,
frozen: bool, # noqa: FBT001
) -> None:
use_bullets = not frozen
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=frozen)
if parent:
prefix = " " * indent + ("- " if use_bullets else "")
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
__all__ = [
"render_text",
]
+123
View File
@@ -0,0 +1,123 @@
from __future__ import annotations
import sys
from collections import defaultdict
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from pipenv.vendor.pipdeptree._models.package import Package
from ._cli import Options
from ._models import DistPackage, PackageDAG, ReqPackage
def validate(args: Options, is_text_output: bool, tree: PackageDAG) -> int: # noqa: FBT001
# Before any reversing or filtering, show warnings to console, about possibly conflicting or cyclic deps if found
# and warnings are enabled (i.e. only if output is to be printed to console)
if is_text_output and args.warn != "silence":
conflicts = conflicting_deps(tree)
if conflicts:
render_conflicts_text(conflicts)
print("-" * 72, file=sys.stderr) # noqa: T201
cycles = cyclic_deps(tree)
if cycles:
render_cycles_text(cycles)
print("-" * 72, file=sys.stderr) # noqa: T201
if args.warn == "fail" and (conflicts or cycles):
return 1
return 0
def conflicting_deps(tree: PackageDAG) -> dict[DistPackage, list[ReqPackage]]:
"""Return dependencies which are not present or conflict with the requirements of other packages.
e.g. will warn if pkg1 requires pkg2==2.0 and pkg2==1.0 is installed
:param tree: the requirements tree (dict)
:returns: dict of DistPackage -> list of unsatisfied/unknown ReqPackage
:rtype: dict
"""
conflicting = defaultdict(list)
for package, requires in tree.items():
for req in requires:
if req.is_conflicting():
conflicting[package].append(req)
return conflicting
def render_conflicts_text(conflicts: dict[DistPackage, list[ReqPackage]]) -> None:
if conflicts:
print("Warning!!! Possibly conflicting dependencies found:", file=sys.stderr) # noqa: T201
# Enforce alphabetical order when listing conflicts
pkgs = sorted(conflicts.keys())
for p in pkgs:
pkg = p.render_as_root(frozen=False)
print(f"* {pkg}", file=sys.stderr) # noqa: T201
for req in conflicts[p]:
req_str = req.render_as_branch(frozen=False)
print(f" - {req_str}", file=sys.stderr) # noqa: T201
def cyclic_deps(tree: PackageDAG) -> list[list[Package]]:
"""Return cyclic dependencies as list of lists.
:param tree: package tree/dag
:returns: list of lists, where each list represents a cycle
"""
def dfs(root: DistPackage, current: Package, visited: set[str], cdeps: list[Package]) -> bool:
if current.key not in visited:
visited.add(current.key)
current_dist = tree.get_node_as_parent(current.key)
if not current_dist:
return False
reqs = tree.get(current_dist)
if not reqs:
return False
for req in reqs:
if dfs(root, req, visited, cdeps):
cdeps.append(current)
return True
elif current.key == root.key:
cdeps.append(current)
return True
return False
cycles: list[list[Package]] = []
for p in tree:
cdeps: list[Package] = []
visited: set[str] = set()
if dfs(p, p, visited, cdeps):
cdeps.reverse()
cycles.append(cdeps)
return cycles
def render_cycles_text(cycles: list[list[Package]]) -> None:
if cycles:
print("Warning!! Cyclic dependencies found:", file=sys.stderr) # noqa: T201
# 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)
for cycle in cycles:
print("*", end=" ", file=sys.stderr) # noqa: T201
size = len(cycle) - 1
for idx, pkg in enumerate(cycle):
if idx == size:
print(f"{pkg.project_name}", end="", file=sys.stderr) # noqa: T201
else:
print(f"{pkg.project_name} =>", end=" ", file=sys.stderr) # noqa: T201
print(file=sys.stderr) # noqa: T201
__all__ = [
"validate",
]
View File
+14 -2
View File
@@ -1,4 +1,16 @@
# file generated by setuptools_scm
# don't change, don't track in version control
__version__ = version = '2.8.0'
__version_tuple__ = version_tuple = (2, 8, 0)
TYPE_CHECKING = False
if TYPE_CHECKING:
from typing import Tuple, Union
VERSION_TUPLE = Tuple[Union[int, str], ...]
else:
VERSION_TUPLE = object
version: str
__version__: str
__version_tuple__: VERSION_TUPLE
version_tuple: VERSION_TUPLE
__version__ = version = '2.13.1'
__version_tuple__ = version_tuple = (2, 13, 1)
+1 -1
View File
@@ -5,7 +5,7 @@ dparse==0.6.3
markupsafe==2.1.2
pep517==0.13.0
pexpect==4.8.0
pipdeptree==2.8.0
pipdeptree==2.13.1
plette==0.4.4
ptyprocess==0.7.0
pydantic==1.10.10
@@ -1,10 +1,18 @@
diff --git a/pipenv/vendor/pipdeptree/__main__.py b/pipenv/vendor/pipdeptree/__main__.py
index 85cca3c..a002019 100644
index cb48791..d1c24f5 100644
--- a/pipenv/vendor/pipdeptree/__main__.py
+++ b/pipenv/vendor/pipdeptree/__main__.py
@@ -1,5 +1,13 @@
@@ -1,6 +1,7 @@
"""The main entry point used for CLI."""
from __future__ import annotations
+import os
import sys
from typing import Sequence
@@ -12,6 +13,13 @@ from pipdeptree._render import render
from pipenv.vendor.pipdeptree._validate import validate
+pardir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+# for finding pipdeptree itself
@@ -13,6 +21,6 @@ index 85cca3c..a002019 100644
+sys.path.append(os.path.dirname(os.path.dirname(pardir)))
+
+
from pipenv.vendor.pipdeptree import main
if __name__ == "__main__":
def main(args: Sequence[str] | None = None) -> None | int:
"""CLI - The main function called as entry point."""
options = get_options(args)