"""Class and functions dealing with dependencies between distributions. This module provides a DependencyGraph class to represent the dependencies between distributions. Auxiliary functions can generate a graph, find reverse dependencies, and print a graph in DOT format. """ import sys from io import StringIO from packaging.errors import PackagingError from packaging.version import VersionPredicate, IrrationalVersionError __all__ = ['DependencyGraph', 'generate_graph', 'dependent_dists', 'graph_to_dot'] class DependencyGraph: """ Represents a dependency graph between distributions. The dependency relationships are stored in an ``adjacency_list`` that maps distributions to a list of ``(other, label)`` tuples where ``other`` is a distribution and the edge is labeled with ``label`` (i.e. the version specifier, if such was provided). Also, for more efficient traversal, for every distribution ``x``, a list of predecessors is kept in ``reverse_list[x]``. An edge from distribution ``a`` to distribution ``b`` means that ``a`` depends on ``b``. If any missing dependencies are found, they are stored in ``missing``, which is a dictionary that maps distributions to a list of requirements that were not provided by any other distributions. """ def __init__(self): self.adjacency_list = {} self.reverse_list = {} self.missing = {} def add_distribution(self, distribution): """Add the *distribution* to the graph. :type distribution: :class:`packaging.database.Distribution` or :class:`packaging.database.EggInfoDistribution` """ self.adjacency_list[distribution] = [] self.reverse_list[distribution] = [] self.missing[distribution] = [] def add_edge(self, x, y, label=None): """Add an edge from distribution *x* to distribution *y* with the given *label*. :type x: :class:`packaging.database.Distribution` or :class:`packaging.database.EggInfoDistribution` :type y: :class:`packaging.database.Distribution` or :class:`packaging.database.EggInfoDistribution` :type label: ``str`` or ``None`` """ self.adjacency_list[x].append((y, label)) # multiple edges are allowed, so be careful if x not in self.reverse_list[y]: self.reverse_list[y].append(x) def add_missing(self, distribution, requirement): """ Add a missing *requirement* for the given *distribution*. :type distribution: :class:`packaging.database.Distribution` or :class:`packaging.database.EggInfoDistribution` :type requirement: ``str`` """ self.missing[distribution].append(requirement) def _repr_dist(self, dist): return '%r %s' % (dist.name, dist.metadata['Version']) def repr_node(self, dist, level=1): """Prints only a subgraph""" output = [] output.append(self._repr_dist(dist)) for other, label in self.adjacency_list[dist]: dist = self._repr_dist(other) if label is not None: dist = '%s [%s]' % (dist, label) output.append(' ' * level + str(dist)) suboutput = self.repr_node(other, level + 1) subs = suboutput.split('\n') output.extend(subs[1:]) return '\n'.join(output) def __repr__(self): """Representation of the graph""" output = [] for dist, adjs in self.adjacency_list.items(): output.append(self.repr_node(dist)) return '\n'.join(output) def graph_to_dot(graph, f, skip_disconnected=True): """Writes a DOT output for the graph to the provided file *f*. If *skip_disconnected* is set to ``True``, then all distributions that are not dependent on any other distribution are skipped. :type f: has to support ``file``-like operations :type skip_disconnected: ``bool`` """ disconnected = [] f.write("digraph dependencies {\n") for dist, adjs in graph.adjacency_list.items(): if len(adjs) == 0 and not skip_disconnected: disconnected.append(dist) for other, label in adjs: if not label is None: f.write('"%s" -> "%s" [label="%s"]\n' % (dist.name, other.name, label)) else: f.write('"%s" -> "%s"\n' % (dist.name, other.name)) if not skip_disconnected and len(disconnected) > 0: f.write('subgraph disconnected {\n') f.write('label = "Disconnected"\n') f.write('bgcolor = red\n') for dist in disconnected: f.write('"%s"' % dist.name) f.write('\n') f.write('}\n') f.write('}\n') def generate_graph(dists): """Generates a dependency graph from the given distributions. :parameter dists: a list of distributions :type dists: list of :class:`packaging.database.Distribution` and :class:`packaging.database.EggInfoDistribution` instances :rtype: a :class:`DependencyGraph` instance """ graph = DependencyGraph() provided = {} # maps names to lists of (version, dist) tuples # first, build the graph and find out the provides for dist in dists: graph.add_distribution(dist) provides = (dist.metadata['Provides-Dist'] + dist.metadata['Provides'] + ['%s (%s)' % (dist.name, dist.metadata['Version'])]) for p in provides: comps = p.strip().rsplit(" ", 1) name = comps[0] version = None if len(comps) == 2: version = comps[1] if len(version) < 3 or version[0] != '(' or version[-1] != ')': raise PackagingError('distribution %r has ill-formed' 'provides field: %r' % (dist.name, p)) version = version[1:-1] # trim off parenthesis if name not in provided: provided[name] = [] provided[name].append((version, dist)) # now make the edges for dist in dists: requires = dist.metadata['Requires-Dist'] + dist.metadata['Requires'] for req in requires: try: predicate = VersionPredicate(req) except IrrationalVersionError: # XXX compat-mode if cannot read the version name = req.split()[0] predicate = VersionPredicate(name) name = predicate.name if name not in provided: graph.add_missing(dist, req) else: matched = False for version, provider in provided[name]: try: match = predicate.match(version) except IrrationalVersionError: # XXX small compat-mode if version.split(' ') == 1: match = True else: match = False if match: graph.add_edge(dist, provider, req) matched = True break if not matched: graph.add_missing(dist, req) return graph def dependent_dists(dists, dist): """Recursively generate a list of distributions from *dists* that are dependent on *dist*. :param dists: a list of distributions :param dist: a distribution, member of *dists* for which we are interested """ if dist not in dists: raise ValueError('given distribution %r is not a member of the list' % dist.name) graph = generate_graph(dists) dep = [dist] # dependent distributions fringe = graph.reverse_list[dist] # list of nodes we should inspect while not len(fringe) == 0: node = fringe.pop() dep.append(node) for prev in graph.reverse_list[node]: if prev not in dep: fringe.append(prev) dep.pop(0) # remove dist from dep, was there to prevent infinite loops return dep def main(): from packaging.database import get_distributions tempout = StringIO() try: old = sys.stderr sys.stderr = tempout try: dists = list(get_distributions(use_egg_info=True)) graph = generate_graph(dists) finally: sys.stderr = old except Exception as e: tempout.seek(0) tempout = tempout.read() print('Could not generate the graph') print(tempout) print(e) sys.exit(1) for dist, reqs in graph.missing.items(): if len(reqs) > 0: print("Warning: Missing dependencies for %r:" % dist.name, ", ".join(reqs)) # XXX replace with argparse if len(sys.argv) == 1: print('Dependency graph:') print(' ', repr(graph).replace('\n', '\n ')) sys.exit(0) elif len(sys.argv) > 1 and sys.argv[1] in ('-d', '--dot'): if len(sys.argv) > 2: filename = sys.argv[2] else: filename = 'depgraph.dot' with open(filename, 'w') as f: graph_to_dot(graph, f, True) tempout.seek(0) tempout = tempout.read() print(tempout) print('Dot file written at %r' % filename) sys.exit(0) else: print('Supported option: -d [filename]') sys.exit(1) if __name__ == '__main__': main()