Skip to content

Reference

This part of the project documentation focuses on an information-oriented approach. Use it as a reference for the technical implementation of the networkx_query project code.

networkx-query.

Evaluator = Callable[[Any], bool] module-attribute

Predicate function.

ParserException

Bases: RuntimeError

Define a parser exception with stack of expression.

Source code in networkx_query/definition.py 
132
133
134
135
136
137
class ParserException(RuntimeError):
    """Define a parser exception with stack of expression."""

    def __init__(self, message, stack):
        super(RuntimeError, self).__init__(message)
        self.stack = stack

PathCriteria

Bases: NamedTuple

Defines path criteria.

Note

With target={}, egde=... you could wrote a criteria based on egde constraint only.

Parameters:

Name Type Description Default
target Dict

target node query constraint. {} means all nodes

 required
edge Dict

optional edge query constraint

 required
Source code in networkx_query/relationship.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class PathCriteria(NamedTuple):
    """Defines path criteria.

    Note:
        With `target={}, egde=...` you could wrote a criteria based on egde constraint only.

    Args:
        target (Dict): target node query constraint.
            `{}` means all nodes
        edge (Dict): optional edge query constraint
    """

    target: Dict
    edge: Optional[Dict] = None

prepare_query(query)

Transform expression query as a function.

Parameters:

Name Type Description Default
query Dict

expression query as dictionary

 required

Returns:

Type Description
Evaluator

evaluator function

Raises:

Type Description
ParserException

if a parse error occurs
Source code in networkx_query/parser.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def prepare_query(query: Dict) -> Evaluator:
    """Transform expression query as a function.

    Arguments:
        query (Dict): expression query as dictionary

    Returns:
        (Evaluator): evaluator function

    Exceptions:
        (ParserException): if a parse error occurs

    """
    return compile_ast(parse(expra=query))

search_direct_relationships(graph, source=None, edge=None, target=None)

Search direct relation ship.

Parameters:

Name Type Description Default
graph Graph

graph instance

 required
source Optional[Dict]

optional source node query constraint

 None
edge Optional[Dict]

optional edge query constraint

 None
target Optional[Dict]

optional target node query constraint

 None

Returns:

Type Description
Iterable[Tuple]

itrable tuple of edge
Source code in networkx_query/relationship.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def search_direct_relationships(
    graph: Graph, source: Optional[Dict] = None, edge: Optional[Dict] = None, target: Optional[Dict] = None
) -> Iterable[Tuple]:
    """Search direct relation ship.

    Arguments:
        graph (Graph): graph instance
        source (Optional[Dict]): optional source node query constraint
        edge (Optional[Dict]): optional edge query constraint
        target (Optional[Dict]): optional target node query constraint

    Returns:
        (Iterable[Tuple]): itrable tuple of edge

    """
    _iterable = search_edges(graph=graph, query=edge) if edge else graph.edges()

    if source:
        _predicate_source = prepare_query(source)
        _iterable = filter(lambda edge: _predicate_source(graph.nodes[edge[0]]), _iterable)

    if target:
        _predicate_target = prepare_query(target)
        _iterable = filter(lambda edge: _predicate_target(graph.nodes[edge[1]]), _iterable)

    return _iterable

search_edges(graph, query)

Search edges in specified graph which match query.

Parameters:

Name Type Description Default
graph Graph

networkx graph instance

 required
query Dict

query expression

 required

Returns:

Type Description
Iterable[Tuple]

results as an iterable of edge identifier (tuple).

Raises:

Type Description
ParserException

if a parse error occurs
Source code in networkx_query/query.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def search_edges(graph: Graph, query: Dict) -> Iterable[Tuple]:
    """Search edges in specified graph which match query.

    Arguments:
        graph (Graph): networkx graph instance
        query (Dict): query expression

    Returns:
        (Iterable[Tuple]): results as an iterable of edge identifier (tuple).

    Exceptions:
        (ParserException): if a parse error occurs

    """
    _predicate = prepare_query(query)
    return map(get_two_first_items, filter(lambda edge: _predicate(edge[2]), graph.edges(data=True)))

search_nodes(graph, query)

Search nodes in specified graph which match query.

Parameters:

Name Type Description Default
graph Graph

networkx graph instance

 required
query Dict

query expression

 required

Returns:

Type Description
Iterable[Any]

results as an iterable of node identifier.

Raises:

Type Description
ParserException

if a parse error occurs
Source code in networkx_query/query.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def search_nodes(graph: Graph, query: Dict) -> Iterable[Any]:
    """Search nodes in specified graph which match query.

    Arguments:
        graph (Graph): networkx graph instance
        query (Dict): query expression

    Returns:
        (Iterable[Any]): results as an iterable of node identifier.

    Exceptions:
        (ParserException): if a parse error occurs

    """
    _predicate = prepare_query(query)
    return map(get_first_item, filter(lambda node: _predicate(node[1]), graph.nodes(data=True)))

search_relationships(graph, source, *nodes)

Search all valid path according specified template.

Parameters:

Name Type Description Default
graph Graph

graph instance

 required
source Dict

source node query constraint

 required
nodes PathCriteria

ordered list of path criteria

 ()

Returns:

Type Description
Iterable[Tuple]

all path which validate specification

Raises:

Type Description
RuntimeError

if a path criteria have no constraint
Source code in networkx_query/relationship.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def search_relationships(graph: Graph, source: Dict, *nodes: PathCriteria) -> Iterable[List]:
    """Search all valid path according specified template.

    Args:
        graph (Graph): graph instance
        source (Dict): source node query constraint
        nodes (PathCriteria): ordered list of path criteria

    Returns:
        (Iterable[Tuple]): all path which validate specification

    Raises:
        (RuntimeError): if a path criteria have no constraint
    """

    if not nodes:
        # cast single edge to path
        for (a, b) in search_direct_relationships(graph=graph, source=source):
            yield [a, b]
        return

    # Validate path template
    node_count = len(nodes)
    assert node_count >= 1

    # build pair of relations ships
    relationships = []

    for i in range(node_count):
        item = set(
            search_direct_relationships(
                graph=graph,
                source=source if i == 0 else nodes[i - 1].target,
                edge=nodes[i].edge,
                target=nodes[i].target,
            )
        )
        relationships.append(item)

        if not len(item):
            raise RuntimeError(f"No valid path from {i}th criteria.")

    # recursive
    def visit(path, node, i):
        for (a, b) in filter(lambda t: t[0] == node, relationships[i]):
            if i + 1 < node_count:
                yield from visit(path=path + [a], node=b, i=i + 1)
            else:
                yield path + [a, b]

    for (a, b) in relationships[0]:
        if node_count == 1:
            yield [a, b]
        else:
            yield from visit(path=[a], node=b, i=1)