diff --git a/swh/graph/client.py b/swh/graph/client.py --- a/swh/graph/client.py +++ b/swh/graph/client.py @@ -15,6 +15,12 @@ return "An unexpected error occurred in the Graph backend: {}".format(self.args) +class GraphArgumentException(Exception): + def __init__(self, *args, response): + super().__init__(*args) + self.response = response + + class RemoteGraphClient(RPCClient): """Client to the Software Heritage Graph.""" @@ -30,6 +36,13 @@ def get_lines(self, endpoint, **kwargs): yield from self.raw_verb_lines("get", endpoint, **kwargs) + def raise_for_status(self, response) -> None: + if response.status_code // 100 == 4: + raise GraphArgumentException( + response.content.decode("ascii"), response=response + ) + super().raise_for_status(response) + # Web API endpoints def stats(self): diff --git a/swh/graph/standalone_client.py b/swh/graph/standalone_client.py new file mode 100644 --- /dev/null +++ b/swh/graph/standalone_client.py @@ -0,0 +1,276 @@ +# Copyright (C) 2021 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 collections +import statistics +from typing import AsyncIterator, Dict, Iterable, Iterator, List, Set, Tuple + +from swh.model.identifiers import ExtendedSWHID + + +class Swhid2NodeDict(collections.UserDict): + def iter_type(self, swhid_type: str) -> Iterator[Tuple[str, str]]: + prefix = "swh:1:{}:".format(swhid_type) + for (swhid, node) in self.items(): + if swhid.startswith(prefix): + yield (swhid, node) + + def __getitem__(self, swhid): + ExtendedSWHID.from_string(swhid) # Raises ValidationError, caught by server + return self.data[swhid] + + +class Node2SwhidDict(collections.UserDict): + def __getitem__(self, key): + try: + return self.data[key] + except KeyError: + # Pretend to be a list + raise IndexError(key) from None + + +class JavaIterator: + def __init__(self, iterator: Iterable): + self.iterator = iter(iterator) + + def nextLong(self): + return next(self.iterator) + + def __getattr__(self, name): + return getattr(self.iterator, name) + + +class StandaloneClient: + """An alternative implementation of :class:`swh.graph.backend.Backend`, + written in pure-python and meant for simulating it in other components' test + cases. + + It is NOT meant to be efficient in any way; only to be a very simple + implementation that provides the same behavior.""" + + def __init__(self, *, nodes: List[str], edges: List[Tuple[str, str]]): + self.graph = Graph(nodes, edges) + + def stats(self) -> Dict: + return { + "counts": { + "nodes": len(self.graph.nodes), + "edges": len(self.graph.forward_edges), + }, + "ratios": { + "compression": 1.0, + "bits_per_edge": 100, + "bits_per_node": 100, + "avg_locality": 0, + }, + "indegree": { + "min": min(map(len, self.graph.backward_edges.values())), + "max": max(map(len, self.graph.backward_edges.values())), + "avg": statistics.mean(map(len, self.graph.backward_edges.values())), + }, + "outdegree": { + "min": min(map(len, self.graph.forward_edges.values())), + "max": max(map(len, self.graph.forward_edges.values())), + "avg": statistics.mean(map(len, self.graph.forward_edges.values())), + }, + } + + def count_neighbors(self, ttype, direction, edges_fmt, src) -> int: + return len(self.graph.get_filtered_neighbors(direction, edges_fmt, src)) + + def count_visit_nodes(self, ttype, direction, edges_fmt, src) -> int: + return len(self.graph.get_subgraph(direction, edges_fmt, src)) + + def count_leaves(self, ttype, direction, edges_fmt, src) -> int: + return len(list(self.leaves(direction, edges_fmt, src))) + + async def simple_traversal(self, ttype, direction, edges_fmt, src, max_edges): + # TODO: max_edges? + if ttype == "visit_nodes": + for node in self.graph.get_subgraph(direction, edges_fmt, src): + yield node + elif ttype == "leaves": + for node in self.leaves(direction, edges_fmt, src): + yield node + else: + assert False, f"unknown ttype {ttype!r}" + + def leaves(self, direction, edges_fmt, src) -> Iterator[str]: + yield from [ + node + for node in self.graph.get_subgraph(direction, edges_fmt, src) + if not self.graph.get_filtered_neighbors(direction, edges_fmt, node) + ] + + async def walk(self, direction, edges_fmt, algo, src, dst) -> AsyncIterator[str]: + # TODO: implement algo="bfs" + if ":" in dst: + match_path = dst.__eq__ + else: + match_path = lambda node: node.startswith(f"swh:1:{dst}:") # noqa + for path in self.graph.iter_paths_dfs(direction, edges_fmt, src): + if match_path(path[-1]): + for node in path: + yield node + + async def random_walk( + self, direction, edges_fmt, retries, src, dst + ) -> AsyncIterator[str]: + async for node in self.walk(direction, edges_fmt, "dfs", src, dst): + yield node + + async def visit_paths( + self, direction, edges_fmt, src, max_edges + ) -> AsyncIterator[List[str]]: + # TODO: max_edges? + for path in self.graph.iter_paths_dfs(direction, edges_fmt, src): + if path[-1] in self.leaves(direction, edges_fmt, src): + yield list(path) + + async def visit_edges( + self, direction, edges_fmt, src, max_edges + ) -> AsyncIterator[Tuple[str, str]]: + if max_edges == 0: + max_edges = None + else: + max_edges -= 1 + edges = list(self.graph.iter_edges_dfs(direction, edges_fmt, src)) + for (from_, to) in edges[:max_edges]: + yield (from_, to) + + +class Graph: + def __init__(self, nodes: List[str], edges: List[Tuple[str, str]]): + self.nodes = nodes + self.forward_edges: Dict[str, List[str]] = {} + self.backward_edges: Dict[str, List[str]] = {} + for (src, dst) in edges: + self.forward_edges.setdefault(src, []).append(dst) + self.backward_edges.setdefault(dst, []).append(src) + + def numNodes(self) -> int: + return len(self.nodes) + + def successors(self, node: str) -> Iterator[str]: + return JavaIterator(self.forward_edges[node]) + + def outdegree(self, node: str) -> int: + return len(self.forward_edges[node]) + + def predecessors(self, node: str) -> Iterator[str]: + return JavaIterator(self.backward_edges[node]) + + def indegree(self, node: str) -> int: + return len(self.backward_edges[node]) + + def get_filtered_neighbors( + self, direction: str, edges_fmt: str, src: str + ) -> Set[str]: + if direction == "forward": + edges = self.forward_edges + elif direction == "backward": + edges = self.backward_edges + else: + assert False, f"unknown direction {direction!r}" + + neighbors = edges.get(src, []) + + if edges_fmt == "*": + return set(neighbors) + else: + filtered_neighbors: Set[str] = set() + for edges_fmt_item in edges_fmt.split(","): + (src_fmt, dst_fmt) = edges_fmt_item.split(":") + if src_fmt != "*" and not src.startswith(f"swh:1:{src_fmt}:"): + continue + if dst_fmt == "*": + filtered_neighbors.update(neighbors) + else: + prefix = f"swh:1:{dst_fmt}:" + filtered_neighbors.update( + n for n in neighbors if n.startswith(prefix) + ) + return filtered_neighbors + + def get_subgraph(self, direction: str, edges_fmt: str, src: str) -> Set[str]: + seen = set() + to_visit = {src} + while to_visit: + node = to_visit.pop() + seen.add(node) + neighbors = set(self.get_filtered_neighbors(direction, edges_fmt, node)) + new_nodes = neighbors - seen + to_visit.update(new_nodes) + + return seen + + def iter_paths_dfs( + self, direction: str, edges_fmt: str, src: str + ) -> Iterator[Tuple[str, ...]]: + for (path, node) in DfsSubgraphIterator(self, direction, edges_fmt, src): + yield path + (node,) + + def iter_edges_dfs( + self, direction: str, edges_fmt: str, src: str + ) -> Iterator[Tuple[str, ...]]: + for (path, node) in DfsSubgraphIterator(self, direction, edges_fmt, src): + if len(path) > 0: + yield (path[-1], node) + + +class SubgraphIterator(Iterator[Tuple[Tuple[str, ...], str]]): + def __init__(self, graph: Graph, direction: str, edges_fmt: str, src: str): + self.graph = graph + self.direction = direction + self.edges_fmt = edges_fmt + self.seen: Set[str] = set() + self.src = src + + def more_work(self) -> bool: + raise NotImplementedError() + + def pop(self) -> Tuple[Tuple[str, ...], str]: + raise NotImplementedError() + + def push(self, new_path: Tuple[str, ...], neighbor: str) -> None: + raise NotImplementedError() + + def __next__(self) -> Tuple[Tuple[str, ...], str]: + # Stores (path, next_node) + if not self.more_work(): + raise StopIteration() + + (path, node) = self.pop() + + new_path = path + (node,) + + if node not in self.seen: + neighbors = self.graph.get_filtered_neighbors( + self.direction, self.edges_fmt, node + ) + + # We want to visit the first neighbor first, and to_visit is a stack; + # so we need to reversed() the list of neighbors to get it on top + # of the stack. + for neighbor in reversed(list(neighbors)): + self.push(new_path, neighbor) + + self.seen.add(node) + return (path, node) + + +class DfsSubgraphIterator(SubgraphIterator): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.to_visit: List[Tuple[Tuple[str, ...], str]] = [((), self.src)] + + def more_work(self) -> bool: + return bool(self.to_visit) + + def pop(self) -> Tuple[Tuple[str, ...], str]: + return self.to_visit.pop() + + def push(self, new_path: Tuple[str, ...], neighbor: str) -> None: + self.to_visit.append((new_path, neighbor)) diff --git a/swh/graph/tests/test_api_client.py b/swh/graph/tests/test_api_client.py --- a/swh/graph/tests/test_api_client.py +++ b/swh/graph/tests/test_api_client.py @@ -2,6 +2,7 @@ from pytest import raises from swh.core.api import RemoteException +from swh.graph.client import GraphArgumentException def test_stats(graph_client): @@ -288,17 +289,17 @@ def test_param_validation(graph_client): - with raises(RemoteException) as exc_info: # SWHID not found + with raises(GraphArgumentException) as exc_info: # SWHID not found list(graph_client.leaves("swh:1:ori:fff0000000000000000000000000000000000021")) assert exc_info.value.response.status_code == 404 - with raises(RemoteException) as exc_info: # malformed SWHID + with raises(GraphArgumentException) as exc_info: # malformed SWHID list( graph_client.neighbors("swh:1:ori:fff000000zzzzzz0000000000000000000000021") ) assert exc_info.value.response.status_code == 400 - with raises(RemoteException) as exc_info: # malformed edge specificaiton + with raises(GraphArgumentException) as exc_info: # malformed edge specificaiton list( graph_client.visit_nodes( "swh:1:dir:0000000000000000000000000000000000000016", @@ -308,7 +309,7 @@ ) assert exc_info.value.response.status_code == 400 - with raises(RemoteException) as exc_info: # malformed direction + with raises(GraphArgumentException) as exc_info: # malformed direction list( graph_client.visit_nodes( "swh:1:dir:0000000000000000000000000000000000000016",