Source code for swh.graph.graph

# Copyright (C) 2019  The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information

import asyncio
import contextlib
import functools

from swh.graph.backend import Backend
from import KIND_TO_SHAPE, dot_to_svg, graph_dot

    "ori": "/origin/{}",
    "snp": "/snapshot/{}",
    "rel": "/release/{}",
    "rev": "/revision/{}",
    "dir": "/directory/{}",
    "cnt": "/content/sha1_git:{}/",

[docs]def call_async_gen(generator, *args, **kwargs): loop = asyncio.get_event_loop() it = generator(*args, **kwargs).__aiter__() while True: try: res = loop.run_until_complete(it.__anext__()) yield res except StopAsyncIteration: break
[docs]class Neighbors: """Neighbor iterator with custom O(1) length method""" def __init__(self, graph, iterator, length_func): self.graph = graph self.iterator = iterator self.length_func = length_func def __iter__(self): return self def __next__(self): succ = self.iterator.nextLong() if succ == -1: raise StopIteration return GraphNode(self.graph, succ) def __len__(self): return self.length_func()
[docs]class GraphNode: """Node in the SWH graph""" def __init__(self, graph, node_id): self.graph = graph = node_id
[docs] def children(self): return Neighbors( self.graph, self.graph.java_graph.successors(, lambda: self.graph.java_graph.outdegree(, )
[docs] def parents(self): return Neighbors( self.graph, self.graph.java_graph.predecessors(, lambda: self.graph.java_graph.indegree(, )
[docs] def simple_traversal( self, ttype, direction="forward", edges="*", max_edges=0, return_types="*" ): for node in call_async_gen( self.graph.backend.simple_traversal, ttype, direction, edges,, max_edges, return_types, ): yield self.graph[node]
[docs] def leaves(self, *args, **kwargs): yield from self.simple_traversal("leaves", *args, **kwargs)
[docs] def visit_nodes(self, *args, **kwargs): yield from self.simple_traversal("visit_nodes", *args, **kwargs)
[docs] def visit_edges(self, direction="forward", edges="*", max_edges=0): for src, dst in call_async_gen( self.graph.backend.visit_edges, direction, edges,, max_edges ): yield (self.graph[src], self.graph[dst])
[docs] def visit_paths(self, direction="forward", edges="*", max_edges=0): for path in call_async_gen( self.graph.backend.visit_paths, direction, edges,, max_edges ): yield [self.graph[node] for node in path]
[docs] def walk(self, dst, direction="forward", edges="*", traversal="dfs"): for node in call_async_gen( self.graph.backend.walk, direction, edges, traversal,, dst ): yield self.graph[node]
def _count(self, ttype, direction="forward", edges="*"): return self.graph.backend.count(ttype, direction, edges, count_leaves = functools.partialmethod(_count, ttype="leaves") count_neighbors = functools.partialmethod(_count, ttype="neighbors") count_visit_nodes = functools.partialmethod(_count, ttype="visit_nodes") @property def swhid(self): return self.graph.node2swhid[] @property def kind(self): return self.swhid.split(":")[2] def __str__(self): return self.swhid def __repr__(self): return "<{}>".format(self.swhid)
[docs] def dot_fragment(self): swh, version, kind, hash = self.swhid.split(":") label = "{}:{}..{}".format(kind, hash[0:2], hash[-2:]) url = BASE_URL + KIND_TO_URL_FRAGMENT[kind].format(hash) shape = KIND_TO_SHAPE[kind] return '{} [label="{}", href="{}", target="_blank", shape="{}"];'.format(, label, url, shape )
def _repr_svg_(self): nodes = [self, *list(self.children()), *list(self.parents())] dot = graph_dot(nodes) svg = dot_to_svg(dot) return svg
[docs]class Graph: def __init__(self, backend, node2swhid, swhid2node): self.backend = backend self.java_graph = backend.entry.get_graph() self.node2swhid = node2swhid self.swhid2node = swhid2node
[docs] def stats(self): return self.backend.stats()
@property def path(self): return self.java_graph.getPath() def __len__(self): return self.java_graph.numNodes() def __getitem__(self, node_id): if isinstance(node_id, int): self.node2swhid[node_id] # check existence return GraphNode(self, node_id) elif isinstance(node_id, str): node_id = self.swhid2node[node_id] return GraphNode(self, node_id) def __iter__(self): for swhid, pos in self.backend.swhid2node: yield self[swhid]
[docs] def iter_prefix(self, prefix): for swhid, pos in self.backend.swhid2node.iter_prefix(prefix): yield self[swhid]
[docs] def iter_type(self, swhid_type): for swhid, pos in self.backend.swhid2node.iter_type(swhid_type): yield self[swhid]
[docs]@contextlib.contextmanager def load(graph_path): with Backend(graph_path) as backend: yield Graph(backend, backend.node2swhid, backend.swhid2node)