aboutsummaryrefslogtreecommitdiffstats
path: root/bsfs/graph/nodes.py
diff options
context:
space:
mode:
Diffstat (limited to 'bsfs/graph/nodes.py')
-rw-r--r--bsfs/graph/nodes.py203
1 files changed, 194 insertions, 9 deletions
diff --git a/bsfs/graph/nodes.py b/bsfs/graph/nodes.py
index 5a93f77..bc71a32 100644
--- a/bsfs/graph/nodes.py
+++ b/bsfs/graph/nodes.py
@@ -5,17 +5,21 @@ A copy of the license is provided with the project.
Author: Matthias Baumgartner, 2022
"""
# imports
+from collections import abc
import time
import typing
# bsfs imports
-from bsfs import schema as _schema
+from bsfs import schema as bsc
from bsfs.namespace import ns
+from bsfs.query import ast, validate
from bsfs.triple_store import TripleStoreBase
from bsfs.utils import errors, URI, typename
# inner-module imports
from . import ac
+from . import result
+from . import walk
# exports
__all__: typing.Sequence[str] = (
@@ -37,7 +41,7 @@ class Nodes():
_user: URI
# node type.
- _node_type: _schema.Node
+ _node_type: bsc.Node
# guids of nodes. Can be empty.
_guids: typing.Set[URI]
@@ -46,13 +50,16 @@ class Nodes():
self,
backend: TripleStoreBase,
user: URI,
- node_type: _schema.Node,
+ node_type: bsc.Node,
guids: typing.Iterable[URI],
):
+ # set main members
self._backend = backend
self._user = user
self._node_type = node_type
self._guids = set(guids)
+ # create helper instances
+ # FIXME: Assumes that the schema does not change while the instance is in use!
self._ac = ac.NullAC(self._backend, self._user)
def __eq__(self, other: typing.Any) -> bool:
@@ -72,7 +79,7 @@ class Nodes():
return f'{typename(self)}({self._node_type}, {self._guids})'
@property
- def node_type(self) -> _schema.Node:
+ def node_type(self) -> bsc.Node:
"""Return the node's type."""
return self._node_type
@@ -81,9 +88,72 @@ class Nodes():
"""Return all node guids."""
return iter(self._guids)
+ @property
+ def schema(self) -> bsc.Schema:
+ """Return the store's local schema."""
+ return self._backend.schema
+
+ def __add__(self, other: typing.Any) -> 'Nodes':
+ """Concatenate guids. Backend, user, and node type must match."""
+ if not isinstance(other, type(self)):
+ return NotImplemented
+ if self._backend != other._backend:
+ raise ValueError(other)
+ if self._user != other._user:
+ raise ValueError(other)
+ if self.node_type != other.node_type:
+ raise ValueError(other)
+ return Nodes(self._backend, self._user, self.node_type, self._guids | other._guids)
+
+ def __or__(self, other: typing.Any) -> 'Nodes':
+ """Concatenate guids. Backend, user, and node type must match."""
+ return self.__add__(other)
+
+ def __sub__(self, other: typing.Any) -> 'Nodes':
+ """Subtract guids. Backend, user, and node type must match."""
+ if not isinstance(other, type(self)):
+ return NotImplemented
+ if self._backend != other._backend:
+ raise ValueError(other)
+ if self._user != other._user:
+ raise ValueError(other)
+ if self.node_type != other.node_type:
+ raise ValueError(other)
+ return Nodes(self._backend, self._user, self.node_type, self._guids - other._guids)
+
+ def __and__(self, other: typing.Any) -> 'Nodes':
+ """Intersect guids. Backend, user, and node type must match."""
+ if not isinstance(other, type(self)):
+ return NotImplemented
+ if self._backend != other._backend:
+ raise ValueError(other)
+ if self._user != other._user:
+ raise ValueError(other)
+ if self.node_type != other.node_type:
+ raise ValueError(other)
+ return Nodes(self._backend, self._user, self.node_type, self._guids & other._guids)
+
+ def __len__(self) -> int:
+ """Return the number of guids."""
+ return len(self._guids)
+
+ def __iter__(self) -> typing.Iterator['Nodes']:
+ """Iterate over individual guids. Returns `Nodes` instances."""
+ return iter(
+ Nodes(self._backend, self._user, self.node_type, {guid})
+ for guid in self._guids
+ )
+
+ def __getattr__(self, name: str):
+ try:
+ return super().__getattr__(name) # type: ignore [misc] # parent has no getattr
+ except AttributeError:
+ pass
+ return walk.Walk(self, walk.Walk.step(self.schema, self.node_type, name))
+
def set(
self,
- pred: URI, # FIXME: URI or _schema.Predicate?
+ pred: URI, # FIXME: URI or bsc.Predicate?
value: typing.Any,
) -> 'Nodes':
"""Set predicate *pred* to *value*."""
@@ -91,7 +161,7 @@ class Nodes():
def set_from_iterable(
self,
- predicate_values: typing.Iterable[typing.Tuple[URI, typing.Any]], # FIXME: URI or _schema.Predicate?
+ predicate_values: typing.Iterable[typing.Tuple[URI, typing.Any]], # FIXME: URI or bsc.Predicate?
) -> 'Nodes':
"""Set mutliple predicate-value pairs at once."""
# TODO: Could group predicate_values by predicate to gain some efficiency
@@ -120,6 +190,120 @@ class Nodes():
return self
+ def get(
+ self,
+ *paths: typing.Union[URI, typing.Iterable[URI]],
+ view: typing.Union[typing.Type[list], typing.Type[dict]] = dict,
+ **view_kwargs,
+ ) -> typing.Any:
+ """Get values or nodes at *paths*.
+ Return an iterator (view=list) or a dict (view=dict) over the results.
+ """
+ # FIXME: user-provided Fetch query AST?
+ # check args
+ if len(paths) == 0:
+ raise AttributeError('expected at least one path, found none')
+ if view not in (dict, list):
+ raise ValueError(f'expected dict or list, found {view}')
+ # process paths: create fetch ast, build name mapping, and find unique paths
+ schema = self.schema
+ statements = set()
+ name2path = {}
+ unique_paths = set() # paths that result in a single (unique) value
+ normpath: typing.Tuple[URI, ...]
+ for idx, path in enumerate(paths):
+ # normalize path
+ if isinstance(path, str):
+ normpath = (URI(path), )
+ elif isinstance(path, abc.Iterable):
+ if not all(isinstance(step, str) for step in path):
+ raise TypeError(path)
+ normpath = tuple(URI(step) for step in path)
+ else:
+ raise TypeError(path)
+ # check path's schema consistency
+ if not all(schema.has_predicate(pred) for pred in normpath):
+ raise errors.ConsistencyError(f'path is not fully covered by the schema: {path}')
+ # check path's uniqueness
+ if all(schema.predicate(pred).unique for pred in normpath):
+ unique_paths.add(path)
+ # fetch tail predicate
+ tail = schema.predicate(normpath[-1])
+ # determine tail ast node type
+ factory = ast.fetch.Node if isinstance(tail.range, bsc.Node) else ast.fetch.Value
+ # assign name
+ name = f'fetch{idx}'
+ name2path[name] = (path, tail)
+ # create tail ast node
+ curr: ast.fetch.FetchExpression = factory(tail.uri, name)
+ # walk towards front
+ hop: URI
+ for hop in normpath[-2::-1]:
+ curr = ast.fetch.Fetch(hop, curr)
+ # add to fetch query
+ statements.add(curr)
+ # aggregate fetch statements
+ if len(statements) == 1:
+ fetch = next(iter(statements))
+ else:
+ fetch = ast.fetch.All(*statements)
+ # add access controls to fetch
+ fetch = self._ac.fetch_read(self.node_type, fetch)
+
+ # compose filter ast
+ filter = ast.filter.IsIn(self.guids) # pylint: disable=redefined-builtin
+ # add access controls to filter
+ filter = self._ac.filter_read(self.node_type, filter)
+
+ # validate queries
+ validate.Filter(self._backend.schema)(self.node_type, filter)
+ validate.Fetch(self._backend.schema)(self.node_type, fetch)
+
+ # process results, convert if need be
+ def triple_iter():
+ # query the backend
+ triples = self._backend.fetch(self.node_type, filter, fetch)
+ # process triples
+ for root, name, raw in triples:
+ # get node
+ node = Nodes(self._backend, self._user, self.node_type, {root})
+ # get path
+ path, tail = name2path[name]
+ # covert raw to value
+ if isinstance(tail.range, bsc.Node):
+ value = Nodes(self._backend, self._user, tail.range, {raw})
+ else:
+ value = raw
+ # emit triple
+ yield node, path, value
+
+ # simplify by default
+ view_kwargs['node'] = view_kwargs.get('node', len(self._guids) != 1)
+ view_kwargs['path'] = view_kwargs.get('path', len(paths) != 1)
+ view_kwargs['value'] = view_kwargs.get('value', False)
+
+ # return results view
+ if view == list:
+ return result.to_list_view(
+ triple_iter(),
+ # aggregation args
+ **view_kwargs,
+ )
+
+ if view == dict:
+ return result.to_dict_view(
+ triple_iter(),
+ # context
+ len(self._guids) == 1,
+ len(paths) == 1,
+ unique_paths,
+ # aggregation args
+ **view_kwargs,
+ )
+
+ raise errors.UnreachableError() # view was already checked
+
+
def __set(self, predicate: URI, value: typing.Any):
"""
"""
@@ -145,7 +329,7 @@ class Nodes():
guids = set(self._ensure_nodes(node_type, guids))
# check value
- if isinstance(pred.range, _schema.Literal):
+ if isinstance(pred.range, bsc.Literal):
# check write permissions on existing nodes
# As long as the user has write permissions, we don't restrict
# the creation or modification of literal values.
@@ -160,8 +344,9 @@ class Nodes():
[value],
)
- elif isinstance(pred.range, _schema.Node):
+ elif isinstance(pred.range, bsc.Node):
# check value type
+ # FIXME: value could be a set of Nodes
if not isinstance(value, Nodes):
raise TypeError(value)
# value's node_type must be a subclass of the predicate's range
@@ -192,7 +377,7 @@ class Nodes():
else:
raise errors.UnreachableError()
- def _ensure_nodes(self, node_type: _schema.Node, guids: typing.Iterable[URI]):
+ def _ensure_nodes(self, node_type: bsc.Node, guids: typing.Iterable[URI]):
"""
"""
# check node existence
generated by cgit v1.2.3 (git 2.25.1) at 2025-08-03 18:06:21 +0000