Merge branch 'mb/fetch' into develop

1 files changed, 55 insertions, 6 deletions

diff --git a/bsfs/triple_store/sparql/sparql.py b/bsfs/triple_store/sparql/sparql.py
index fedd227..dbf9d45 100644
--- a/bsfs/triple_store/sparql/sparql.py
+++ b/bsfs/triple_store/sparql/sparql.py
@@ -5,8 +5,11 @@ A copy of the license is provided with the project.
 Author: Matthias Baumgartner, 2022
 """
 # imports
+import base64
 import itertools
 import typing
+
+# external imports
 import rdflib
 
 # bsfs imports
@@ -16,6 +19,7 @@ from bsfs.query import ast
 from bsfs.utils import errors, URI
 
 # inner-module imports
+from . import parse_fetch
 from . import parse_filter
 from .. import base
 from .distance import DISTANCE_FU
@@ -29,6 +33,8 @@ __all__: typing.Sequence[str] = (
 
 ## code ##
 
+rdflib.term.bind(ns.bsfs.BinaryBlob, bytes, constructor=base64.b64decode)
+
 class _Transaction():
     """Lightweight rdflib transactions for in-memory databases."""
 
@@ -92,13 +98,16 @@ class SparqlStore(base.TripleStoreBase):
     # Filter parser
     _filter_parser: parse_filter.Filter
 
+    # Fetch parser
+    _fetch_parser: parse_fetch.Fetch
+
     def __init__(self):
         super().__init__(None)
         self._graph = rdflib.Graph()
         self._transaction = _Transaction(self._graph)
-        # NOTE: parsing bsfs.query.ast.filter.Has requires xsd:integer.
         self._schema = bsc.Schema(literals={bsc.ROOT_NUMBER.child(ns.xsd.integer)})
         self._filter_parser = parse_filter.Filter(self._graph, self._schema)
+        self._fetch_parser = parse_fetch.Fetch(self._schema)
 
     # NOTE: mypy and pylint complain about the **kwargs not being listed (contrasting super)
     # However, not having it here is clearer since it's explicit that there are no arguments.
@@ -197,17 +206,53 @@ class SparqlStore(base.TripleStoreBase):
         # migrate schema
         self._schema = schema
         self._filter_parser.schema = schema
+        self._fetch_parser.schema = schema
+
+    def fetch(
+            self,
+            node_type: bsc.Node,
+            filter: ast.filter.FilterExpression, # pylint: disable=redefined-builtin
+            fetch: ast.fetch.FetchExpression,
+            ) -> typing.Iterator[typing.Tuple[URI, str, typing.Any]]:
+        if node_type not in self.schema.nodes():
+            raise errors.ConsistencyError(f'{node_type} is not defined in the schema')
+        if not isinstance(filter, ast.filter.FilterExpression):
+            raise TypeError(filter)
+        if not isinstance(fetch, ast.fetch.FetchExpression):
+            raise TypeError(fetch)
+        # compose a query from fetch and filter ast
+        query = self._filter_parser(node_type, filter)
+        query += self._fetch_parser(node_type, fetch)
+        # run query
+        emitted = set()
+        for result in query(self._graph):
+            guid = URI(result[0])
+            for name, raw in zip(query.names, result[1:]):
+                if raw is None: # undefined value
+                    continue
+                if isinstance(raw, rdflib.Literal):
+                    value = raw.value
+                else:
+                    value = URI(raw)
+                # emit triple
+                triple = (guid, name, value)
+                if triple not in emitted: # FIXME: needs a better solution!
+                    emitted.add(triple)
+                    yield guid, name, value
 
     def get(
             self,
             node_type: bsc.Node,
-            query: typing.Optional[ast.filter.FilterExpression] = None,
+            filter: typing.Optional[ast.filter.FilterExpression] = None, # pylint: disable=redefined-builtin
             ) -> typing.Iterator[URI]:
         if node_type not in self.schema.nodes():
             raise errors.ConsistencyError(f'{node_type} is not defined in the schema')
-        if not isinstance(query, ast.filter.FilterExpression):
-            raise TypeError(query)
-        for guid, in self._graph.query(self._filter_parser(node_type, query)):
+        if filter is not None and not isinstance(filter, ast.filter.FilterExpression):
+            raise TypeError(filter)
+        # compose query
+        query = self._filter_parser(node_type, filter)
+        # run query
+        for guid, in query(self._graph):
             yield URI(guid)
 
     def _has_type(self, subject: URI, node_type: bsc.Node) -> bool:
@@ -294,7 +339,11 @@ class SparqlStore(base.TripleStoreBase):
             guid = rdflib.URIRef(guid)
             # convert value
             if isinstance(predicate.range, bsc.Literal):
-                value = rdflib.Literal(value, datatype=rdflib.URIRef(predicate.range.uri))
+                dtype = rdflib.URIRef(predicate.range.uri)
+                if predicate.range <= self.schema.literal(ns.bsfs.BinaryBlob):
+                    dtype = rdflib.URIRef(ns.bsfs.BinaryBlob)
+                    value = base64.b64encode(value)
+                value = rdflib.Literal(value, datatype=dtype)
             elif isinstance(predicate.range, bsc.Node):
                 value = rdflib.URIRef(value)
             else: