swh.graph.http_naive_client module

swh.graph.http_naive_client.check_arguments(f: T) → T

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]

class swh.graph.http_naive_client.NaiveClient(*, nodes: List[CoreSWHID | ExtendedSWHID | str], edges: List[Tuple[CoreSWHID | ExtendedSWHID | str, CoreSWHID | ExtendedSWHID | str]])

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

leaves(src: str, edges: str = '*', direction: str = 'forward', max_edges: int = 0, return_types: str = '*', max_matching_nodes: int = 0) → Iterator[str]

neighbors(src: str, edges: str = '*', direction: str = 'forward', max_edges: int = 0, return_types: str = '*', max_matching_nodes: int = 0) → Iterator[str]

visit_nodes(src: str, edges: str = '*', direction: str = 'forward', max_edges: int = 0, return_types: str = '*', max_matching_nodes: int = 0) → Iterator[str]

visit_edges(src: str, edges: str = '*', direction: str = 'forward', max_edges: int = 0) → Iterator[Tuple[str, str]]

visit_paths(src: str, edges: str = '*', direction: str = 'forward', max_edges: int = 0) → Iterator[List[str]]

walk(src: str, dst: str, edges: str = '*', traversal: str = 'dfs', direction: str = 'forward', limit: int | None = None) → Iterator[str]

random_walk(src: str, dst: str, edges: str = '*', direction: str = 'forward', limit: int | None = None)

count_leaves(src: str, edges: str = '*', direction: str = 'forward', max_matching_nodes: int = 0) → int

count_neighbors(src: str, edges: str = '*', direction: str = 'forward') → int

count_visit_nodes(src: str, edges: str = '*', direction: str = 'forward', max_matching_nodes: int = 0) → int

class swh.graph.http_naive_client.Graph(nodes: List[CoreSWHID | ExtendedSWHID | str], edges: List[Tuple[CoreSWHID | ExtendedSWHID | str, CoreSWHID | ExtendedSWHID | str]])

Bases: object

get_filtered_neighbors(src: str, edges_fmt: str, direction: str, max_edges: int = 0, max_matching_nodes: int = 0) → Set[str]

get_subgraph(src: str, edges_fmt: str, direction: str) → Set[str]

iter_paths_dfs(direction: str, edges_fmt: str, src: str) → Iterator[Tuple[str, ...]]

iter_edges_dfs(direction: str, edges_fmt: str, src: str) → Iterator[Tuple[str, str]]

class swh.graph.http_naive_client.SubgraphIterator(graph: Graph, direction: str, edges_fmt: str, src: str)

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

more_work() → bool

pop() → Tuple[Tuple[str, ...], str]

push(new_path: Tuple[str, ...], neighbor: str) → None

class swh.graph.http_naive_client.DfsSubgraphIterator(*args, **kwargs)

Bases: SubgraphIterator

more_work() → bool

pop() → Tuple[Tuple[str, ...], str]

push(new_path: Tuple[str, ...], neighbor: str) → None