Extend refactoring of RDF processing

Author: nichtich

lib/__init__.py

@@ -3,7 +3,8 @@
 from .mappings import MappingRegistry
 from .errors import ApiError, NotFound, NotAllowed, ServerError
 from .utils import read_json, write_json
-from .rdf import createTripleStore, triple_iterator
+from .rdf import triple_iterator
+from .triplestores import createTripleStore
 from .rdffilter import RDFFilter
 from .validate import validateJSON, ValidationError
 

lib/rdf.py

@@ -1,6 +1,4 @@
-from .triplestores import createTripleStore
 from pyld import jsonld
-from rdflib import URIRef, Literal, BNode
 from .errors import ValidationError
 from .walk import walk
 from .validate import invalidIRI
@@ -18,26 +16,6 @@ def jsonld2nt(doc, context):
     return jsonld.to_rdf(expanded, options={'format': 'application/n-quads'})
 
 
-def sparql_to_rdf(binding):
-    if binding['type'] == 'uri':
-        return URIRef(binding['value'])
-    elif binding['type'] == 'bnode':
-        return BNode(binding['value'])
-    elif binding['type'] == 'literal':
-        if 'datatype' in binding:
-            return Literal(binding['value'], datatype=URIRef(binding['datatype']))
-        elif 'xml:lang' in binding:
-            return Literal(binding['value'], lang=binding['xml:lang'])
-        else:
-            return Literal(binding['value'])
-
-
-def result_to_ttl(data):
-    rows = [dict([(key, sparql_to_rdf(val).n3())
-                 for key, val in row.items()]) for row in data]
-    return "\n".join([f"{row['s']} {row['p']} {row['o']} ." for row in rows])
-
-
 class NullLog:
     def append(self, msg):
         pass

lib/registry.py

@@ -3,7 +3,7 @@
 from shutil import copy, copyfileobj, rmtree
 import urllib
 from .validate import validateJSON
-from .rdf import jsonld2nt, result_to_ttl
+from .rdf import jsonld2nt
 from .rdffilter import RDFFilter
 from .log import Log
 from .errors import ApiError, NotFound, ValidationError
@@ -91,7 +91,7 @@ def replace(self, items):
 
     def update_metadata(self):
         query = f"SELECT * {{ GRAPH <{self.graph}> {{ VALUES (?p) {{(dct:issued)}} ?s ?p ?o }} }}"
-        issued = result_to_ttl(self.sparql.query(query))
+        issued = self.sparql.query(query, "nq")
         metadata = jsonld2nt(self.list(), self.context)
         file = self.stage / f"{self.kind}.ttl"
         with open(file, "w") as f:

lib/triplestores.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from rdflib import URIRef, BNode, Dataset, Graph
+from rdflib import URIRef, Literal, BNode, Dataset, Graph
 from SPARQLWrapper import SPARQLWrapper
 import requests
 import warnings
@@ -26,7 +26,7 @@ class AbstractTripleStore(ABC):
 
     """Must implement SPARQL Query"""
     @abstractmethod
-    def query(self, query):
+    def query(self, query, format):
         pass
 
     """Import"""
@@ -67,10 +67,11 @@ def __client(self, query):
         client.setQuery(self.prefixes + query)
         return client
 
-    def query(self, query):
+    def query(self, query, format='sparql'):
         client = self.__client(query)
         try:
-            return client.queryAndConvert()["results"]["bindings"]
+            result = client.queryAndConvert()["results"]["bindings"]
+            return convert_query_result(result, convert_sparql_term, format)
         except Exception as e:
             raise ServerError(f"SPARQL Query failed: {e}")
 
@@ -97,28 +98,17 @@ class InternalTripleStore(AbstractTripleStore):
     def __init__(self):
         self.ds = Dataset(default_union=True)
 
-    def query(self, query):
-        def term(t):
-            if isinstance(t, URIRef):
-                return {"type": "uri", "value": str(t)}
-            if isinstance(t, BNode):
-                return {"type": "bnode", "value": str(t)}
-            literal = {"type": "literal", "value": str(t)}
-            if t.language:
-                literal["xml:lang"] = t.language
-            if t.datatype:
-                literal["datatype"] = str(t.datatype)
-            return literal
-
+    def query(self, query, format='sparql'):
         def map_row(row):
-            return {str(k): term(v) for k, v in row.items()}
+            return {str(k): convert_rdflib_term(v, format) for k, v in row.items()}
 
         query = self.prefixes + query
 
         # RDFLib raises warning, see <https://github.com/RDFLib/rdflib/issues/3361>
         with warnings.catch_warnings():
             warnings.filterwarnings("ignore", category=DeprecationWarning)
-            return [map_row(row) for row in self.ds.query(query).bindings]
+            result = self.ds.query(query).bindings
+            return convert_query_result(result, convert_rdflib_term, format)
 
     def _update(self, query):
         self.ds.update(self.prefixes + query)
@@ -133,3 +123,50 @@ def store_file(self, graph, file):
         for triple in data:
             graph.add(triple)
         return True
+
+
+def convert_query_result(result, mapper, target):
+    """Convert a SPARQL Query result to target form (sparql, rdflib, n3, nq, ttl)."""
+
+    if target == "nq" or target == "ttl":
+        result = convert_query_result(result, mapper, "n3")
+        return "\n".join([
+            " ".join([row.get(f) for f in ['g', 's', 'p', 'o'] if f in row]) + " ."
+            for row in result])
+
+    return [{str(k): mapper(v, target) for k, v in row.items()} for row in result]
+
+
+def convert_sparql_term(term, format):
+    if format == "rdflib" or format == "n3":
+        if term['type'] == 'uri':
+            term = URIRef(term['value'])
+        elif term['type'] == 'bnode':
+            term = BNode(term['value'])
+        elif term['type'] == 'literal':
+            if 'datatype' in term:
+                term = Literal(term['value'], datatype=URIRef(term['datatype']))
+            elif 'xml:lang' in term:
+                term = Literal(term['value'], lang=term['xml:lang'])
+            else:
+                term = Literal(term['value'])
+        if format == "n3":
+            return term.n3()
+    return term
+
+
+def convert_rdflib_term(term, format):
+    if format == "rdflib":
+        return term
+    if format == "n3":
+        return term.n3()
+    if isinstance(term, URIRef):
+        return {"type": "uri", "value": str(term)}
+    if isinstance(term, BNode):
+        return {"type": "bnode", "value": str(term)}
+    literal = {"type": "literal", "value": str(term)}
+    if term.language:
+        literal["xml:lang"] = term.language
+    if term.datatype:
+        literal["datatype"] = str(term.datatype)
+    return literal

tests/test_api.py

@@ -57,8 +57,10 @@ def expect_error(client, method, path, json=None, error=None, code=400):
 def count_graphs(sparql):
     query = "SELECT ?g (count(*) as ?t) { GRAPH ?g {?s ?p ?o} } GROUP BY ?g"
     graphs = {}
+
     for row in sparql.query(query):
-        graphs[row['g']['value']] = int(row['t']['value'])
+        if "g" in row:  # <https://github.com/RDFLib/rdflib/issues/3382>
+            graphs[row['g']['value']] = int(row['t']['value'])
     return graphs
 
 
@@ -124,14 +126,15 @@ def test_terminology(client):
     fail("GET", "/terminology/18274", code=404)
     fail("GET", "/terminology/18274/stage/", code=404)
 
+    # register terminology from BARTOC
+    assert count_graphs(sparql) == {}
     with patch('requests.get', new=mock_requests_get):
-
-        # register terminology, get afterwards
         assert client.put("/terminology/18274").status_code == 200
         assert client.get("/terminology/18274").status_code == 200
 
         # try to register non-existing terminology
         fail("PUT", "/terminology/0", code=404)
+    assert count_graphs(sparql) == {'https://graph.nfdi4objects.net/terminology/': 12}
 
     assert client.get("/terminology/18274/stage/").status_code == 200
     fail("GET", "/terminology/18274/stage/terminology-18274.nt", code=404)
@@ -169,6 +172,10 @@ def test_terminology(client):
     assert client.get('/terminology/18274/receive').status_code == 200
     assert client.get("/terminology/18274/stage/terminology-18274.nt").status_code == 200
 
+    # FIXME: Here we get too many bnodes in InternalTripleStore
+    # print(sparql.query("SELECT * { GRAPH ?g { ?s ?p ?o } }", "ttl"))
+    # assert count_graphs(sparql) == { 'https://graph.nfdi4objects.net/terminology/': 99 } # 12
+
     # load terminology data and check log
     fail("GET", '/terminology/18274/load', code=404)
     assert client.post('/terminology/18274/load').status_code == 200

tests/test_collections.py

@@ -1,8 +1,7 @@
 # Unit test
 import pytest
 import tempfile
-from lib import CollectionRegistry
-from lib.rdf import createTripleStore
+from lib import CollectionRegistry, createTripleStore
 
 
 @pytest.fixture

tests/test_rdf.py

@@ -1,6 +1,5 @@
 # Unit test
 from lib import triple_iterator, RDFFilter, ValidationError
-from lib.triplestores import createTripleStore
 
 
 def parse(source, filter=None, unique=False):
@@ -20,7 +19,7 @@ def fail(file, error):
         assert e.to_dict() == error
 
 
-def _test_parsing():
+def test_parsing():
     assert len(parse("tests/skos.rdf")) == 377
     assert len(parse("tests/rdf.zip")) == 4
     assert len(parse("tests/iri.ttl")) == 1
@@ -63,27 +62,10 @@ def _test_parsing():
     })
 
 
-def _test_filter():
+def test_filter():
     triples = parse("tests/filter.ttl")
     assert len(triples) == 7
 
     filter = RDFFilter(disallow_subject_ns=("http://www.cidoc-crm.org/cidoc-crm/"))
     triples = parse("tests/filter.ttl", filter, True)
     assert len(triples) == 3
-
-
-def test_store():
-    store = createTripleStore()
-
-    store.insert('http://example.org/', '_:b1 dct:title "foo"')
-    assert store.query("SELECT * { ?s ?p ?o }") == [{
-        's': {'type': 'bnode', 'value': 'b1'},
-        'p': {'type': 'uri', 'value': 'http://purl.org/dc/terms/title'},
-        'o': {'type': 'literal', 'value': 'foo'}
-    }]
-
-    return
-
-    store.store_file('http://example.org/1', "tests/filter.ttl")
-    query = "SELECT * { GRAPH <http://example.org/1> { ?s ?b ?o } }"
-    assert len(store.query(query)) == 6

tests/test_triplestores.py

@@ -0,0 +1,29 @@
+# Unit test
+from lib.triplestores import createTripleStore
+from rdflib import URIRef, Literal, BNode
+
+
+def test_store():
+    store = createTripleStore()
+
+    store.insert('http://example.org/', '_:b1 dct:title "foo"')
+    assert store.query("SELECT * { ?s ?p ?o }") == [{
+        's': {'type': 'bnode', 'value': 'b1'},
+        'p': {'type': 'uri', 'value': 'http://purl.org/dc/terms/title'},
+        'o': {'type': 'literal', 'value': 'foo'}
+    }]
+    assert store.query("SELECT * { ?s ?p ?o }", "rdflib") == [{
+        's': BNode('b1'),
+        'p': URIRef('http://purl.org/dc/terms/title'),
+        'o': Literal("foo")
+    }]
+    assert store.query("SELECT * { ?s ?p ?o }", "n3") == [{
+        's': '_:b1',
+        'p': '<http://purl.org/dc/terms/title>',
+        'o': '"foo"'
+    }]
+    assert store.query("SELECT * { ?s ?p ?o }", "nq") == '_:b1 <http://purl.org/dc/terms/title> "foo" .'
+
+    store.store_file('http://example.org/1', "tests/filter.ttl")
+    query = "SELECT * { GRAPH <http://example.org/1> { ?s ?b ?o } }"
+    assert len(store.query(query)) == 6