1
2
3
4
5
6
7
8
9
10
11
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
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
"""
Part of the BlackStar filesystem (bsfs) module.
A copy of the license is provided with the project.
Author: Matthias Baumgartner, 2022
"""
# imports
from collections import defaultdict
import typing
# bsfs imports
from bsfs.utils import URI
# exports
__all__: typing.Sequence[str] = (
'to_list_view',
'to_dict_view',
)
## code ##
# FIXME: node, path, value seem counter-intuitive:
# node.get(..., node=True) removes the node part.
# wouldn't it make more sense if node=True keeps the node part
# and node=False drops it?
def to_list_view(
triples,
# aggregators
node: bool,
path: bool,
value: bool, # pylint: disable=unused-argument
):
"""Return an iterator over results.
Dependent on the *node*, *path*, and *value* flags,
the respective component is omitted.
"""
if not node and not path:
return iter(val for _, _, val in triples)
if not node:
return iter((pred, val) for _, pred, val in triples)
if not path:
return iter((subj, val) for subj, _, val in triples)
return iter((subj, pred, val) for subj, pred, val in triples)
def to_dict_view(
triples,
# context
one_node: bool,
one_path: bool,
unique_paths: typing.Set[typing.Union[URI, typing.Iterable[URI]]],
# aggregators
node: bool,
path: bool,
value: bool,
default: typing.Optional[typing.Any] = None,
) -> typing.Any:
"""Return a dict of results.
Note that triples are materialized to create this view.
The returned structure depends on the *node*, *path*, and *value* flags.
If all flags are set to False, returns a dict(node -> dict(path -> set(values))).
Setting a flag to true omits or simplifies the respective component (if possible).
"""
# NOTE: To create a dict, we need to materialize or make further assumptions
# (e.g., sorted in a specific order).
data: typing.Any # disable type checks on data since it's very flexibly typed.
# FIXME: type of data can be overwritten later on (if value)
if not node and not path:
data = set()
elif node ^ path:
data = defaultdict(set)
else:
data = defaultdict(lambda: defaultdict(set))
for subj, pred, val in triples:
unique = pred in unique_paths
if not node and not path:
if not value and unique and one_node and one_path:
return val
data.add(val)
elif not node:
# remove node from result, group by predicate
if not value and unique and one_node:
data[pred] = val
else:
data[pred].add(val)
elif not path:
# remove predicate from result, group by node
if not value and unique and one_path:
data[subj] = val
else:
data[subj].add(val)
else:
if not value and unique:
data[subj][pred] = val
else:
data[subj][pred].add(val)
# FIXME: Combine multiple Nodes instances into one?
# convert defaultdict to ordinary dict
# pylint: disable=too-many-boolean-expressions
if not node and not path and not value \
and len(unique_paths) > 0 and one_node and one_path \
and len(data) == 0:
return default
# pylint: enable=too-many-boolean-expressions
if not node and not path:
return data
if node ^ path:
return dict(data)
return {key: dict(val) for key, val in data.items()}
## EOF ##
|
