swh.graph.http_naive_client module

swh.graph.http_naive_client.check_arguments(f: T) T[source]

Decorator for generic argument checking for methods of NaiveClient. Checks src is a valid and known SWHID, and edges has the right format.

swh.graph.http_naive_client.filter_node_types(node_types: str, nodes: Iterable[str]) Iterator[str][source]
class swh.graph.http_naive_client.NaiveClient(*, nodes: List[Union[CoreSWHID, ExtendedSWHID, str]], edges: List[Tuple[Union[CoreSWHID, ExtendedSWHID, str], Union[CoreSWHID, ExtendedSWHID, str]]])[source]

Bases: object

An alternative implementation of the graph server, written in pure-python and meant for simulating it in other components’ test cases; constructed from a list of nodes and (directed) edges, both represented as SWHIDs.

It is NOT meant to be efficient in any way; only to be a very simple implementation that provides the same behavior.

>>> nodes = [
...     "swh:1:rev:1111111111111111111111111111111111111111",
...     "swh:1:rev:2222222222222222222222222222222222222222",
...     "swh:1:rev:3333333333333333333333333333333333333333",
... ]
>>> edges = [
...     (
...         "swh:1:rev:1111111111111111111111111111111111111111",
...         "swh:1:rev:2222222222222222222222222222222222222222",
...     ),
...     (
...         "swh:1:rev:2222222222222222222222222222222222222222",
...         "swh:1:rev:3333333333333333333333333333333333333333",
...     ),
... ]
>>> c = NaiveClient(nodes=nodes, edges=edges)
>>> list(c.leaves("swh:1:rev:1111111111111111111111111111111111111111"))
['swh:1:rev:3333333333333333333333333333333333333333']
stats() Dict[source]
leaves(src: str, edges: str = '*', direction: str = 'forward', max_edges: int = 0, return_types: str = '*', max_matching_nodes: int = 0) Iterator[str][source]
neighbors(src: str, edges: str = '*', direction: str = 'forward', max_edges: int = 0, return_types: str = '*') Iterator[str][source]
visit_nodes(src: str, edges: str = '*', direction: str = 'forward', max_edges: int = 0, return_types: str = '*', max_matching_nodes: int = 0) Iterator[str][source]
visit_edges(src: str, edges: str = '*', direction: str = 'forward', max_edges: int = 0) Iterator[Tuple[str, str]][source]
visit_paths(src: str, edges: str = '*', direction: str = 'forward', max_edges: int = 0) Iterator[List[str]][source]
walk(src: str, dst: str, edges: str = '*', traversal: str = 'dfs', direction: str = 'forward', limit: Optional[int] = None) Iterator[str][source]
random_walk(src: str, dst: str, edges: str = '*', direction: str = 'forward', limit: Optional[int] = None)[source]
count_leaves(src: str, edges: str = '*', direction: str = 'forward', max_matching_nodes: int = 0) int[source]
count_neighbors(src: str, edges: str = '*', direction: str = 'forward') int[source]
count_visit_nodes(src: str, edges: str = '*', direction: str = 'forward', max_matching_nodes: int = 0) int[source]
class swh.graph.http_naive_client.Graph(nodes: List[Union[CoreSWHID, ExtendedSWHID, str]], edges: List[Tuple[Union[CoreSWHID, ExtendedSWHID, str], Union[CoreSWHID, ExtendedSWHID, str]]])[source]

Bases: object

get_filtered_neighbors(src: str, edges_fmt: str, direction: str) Set[str][source]
get_subgraph(src: str, edges_fmt: str, direction: str) Set[str][source]
iter_paths_dfs(direction: str, edges_fmt: str, src: str) Iterator[Tuple[str, ...]][source]
iter_edges_dfs(direction: str, edges_fmt: str, src: str) Iterator[Tuple[str, str]][source]
class swh.graph.http_naive_client.SubgraphIterator(graph: Graph, direction: str, edges_fmt: str, src: str)[source]

Bases: Iterator[Tuple[Tuple[str, …], str]]

more_work() bool[source]
pop() Tuple[Tuple[str, ...], str][source]
push(new_path: Tuple[str, ...], neighbor: str) None[source]
class swh.graph.http_naive_client.DfsSubgraphIterator(*args, **kwargs)[source]

Bases: SubgraphIterator

more_work() bool[source]
pop() Tuple[Tuple[str, ...], str][source]
push(new_path: Tuple[str, ...], neighbor: str) None[source]