# 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 ##