Add ability to specify SPARQL queries on classes

spec_parser/mdparsing.py

@@ -4,6 +4,7 @@
 
 import logging
 import re
+import textwrap
 
 logger = logging.getLogger(__name__)
 
@@ -81,24 +82,53 @@ def load(self, content):
 
 class NestedListSection(Section):
     RE_EXTRACT_TOP_LEVEL = re.compile(r"-\s+((\w|/)+)")
-    RE_EXTRACT_KEY_VALUE = re.compile(r"\s+-\s+(\w+):\s+(.+)")
+    RE_EXTRACT_KEY_VALUE = re.compile(r"(\s+-\s+)(\w+):((\s+.+)?)")
 
     def load(self, content):
         self.content = content
         self.ikv = dict()
+        item = None
+        key = None
+        value = None
+        indent = None
+
+        def add_value():
+            nonlocal item
+            nonlocal key
+            nonlocal value
+            if not key:
+                return
+
+            self.ikv[item][key] = textwrap.dedent(value)
+            key = None
+
         for i,l in enumerate(content.splitlines()):
+            if not l.rstrip():
+                continue
+
             if l.startswith("-"):
                 m = re.fullmatch(self.RE_EXTRACT_TOP_LEVEL, l)
                 if m is None:
                     logger.error(self._fmt_err_msg("Top-level nested list parsing error", i+1, l))
                 else:
+                    add_value()
                     item = m.group(1)
                     self.ikv[item] = dict()
+            elif item is None:
+                logger.error(self._fmt_err_msg("Top-level nested list parsing error", i+1, l))
             else:
                 m = re.fullmatch(self.RE_EXTRACT_KEY_VALUE, l)
                 if m is None:
-                    logger.error(self._fmt_err_msg("Nested list parsing error", i+1, l))
+                    if indent and l.startswith(" " * len(indent)):
+                        if value:
+                            value += "\n"
+                        value += l
+                    else:
+                        logger.error(self._fmt_err_msg("Nested list parsing error", i+1, l))
                 else:
-                    key = m.group(1)
-                    val = m.group(2).strip()
-                    self.ikv[item][key] = val
+                    add_value()
+                    indent = m.group(1)
+                    key = m.group(2)
+                    value = m.group(3).strip()
+
+        add_value()

spec_parser/model.py

@@ -264,6 +264,12 @@ def __init__(self, fname, ns):
         else:
             self.ext_prop_restrs = dict()
 
+        if "SPARQL" in sf.sections:
+            s = NestedListSection(sf.sections["SPARQL"], filename=self.fqname, context="sparql")
+            self.sparql = s.ikv
+        else:
+            self.sparql = dict()
+
         # checks
         assert self.name == self.metadata["name"], f"Class name {self.name} does not match metadata {self.metadata['name']}"
         for p in self.metadata:
@@ -320,6 +326,12 @@ def __init__(self, fname, ns):
         s = SingleListSection(sf.sections["Metadata"], filename=self.fqname, context="metadata")
         self.metadata = s.kv
 
+        if "SPARQL" in sf.sections:
+            s = NestedListSection(sf.sections["SPARQL"], filename=self.fqname, context="sparql")
+            self.sparql = s.ikv
+        else:
+            self.sparql = dict()
+
         # checks
         assert self.name == self.metadata["name"], f"Property name {self.name} does not match metadata {self.metadata['name']}"
         for p in self.metadata:
@@ -355,6 +367,12 @@ def __init__(self, fname, ns):
         s = SingleListSection(sf.sections["Entries"], filename=self.fqname, context="entries")
         self.entries = s.kv
 
+        if "SPARQL" in sf.sections:
+            s = NestedListSection(sf.sections["SPARQL"], filename=self.fqname, context="sparql")
+            self.sparql = s.ikv
+        else:
+            self.sparql = dict()
+
         # checks
         assert self.name == self.metadata["name"], f"Vocabulary name {self.name} does not match metadata {self.metadata['name']}"
         for p in self.metadata:

spec_parser/rdf.py

@@ -83,6 +83,16 @@ def get_parent(model, c):
     return None
 
 
+def add_sparql(g, target, sparql):
+    snode = BNode()
+    g.add((snode, RDF.type, SH.SPARQLConstraint))
+    g.add((snode, SH.message, Literal(sparql["message"])))
+
+    query = f"BASE <{URI_BASE}>\n" + sparql["query"]
+    g.add((snode, SH.select, Literal(query)))
+
+    g.add((target, SH.sparql, snode))
+
 def gen_rdf_classes(model, g):
     for c in model.classes.values():
         node = URIRef(c.iri)
@@ -110,6 +120,9 @@ def gen_rdf_classes(model, g):
         else:
             g.add((node, SH.nodeKind, SH.BlankNodeOrIRI))
 
+        for s in c.sparql.values():
+            add_sparql(g, node, s)
+
         if c.properties:
             g.add((node, RDF.type, SH.NodeShape))
             for p in c.properties:
@@ -119,6 +132,10 @@ def gen_rdf_classes(model, g):
                 bnode = BNode()
                 g.add((node, SH.property, bnode))
                 prop = model.properties[fqprop]
+
+                for s in prop.sparql.values():
+                    add_sparql(g, node, s)
+
                 g.add((bnode, SH.path, URIRef(prop.iri)))
                 prop_rng = prop.metadata["Range"]
                 if ":" not in prop_rng:
@@ -169,6 +186,8 @@ def gen_rdf_classes(model, g):
                     for e in dt.entries:
                         lst.append(URIRef(dt.iri + "/" + e))
                     g.add((bnode, SH["in"], lst.uri))
+                    for s in dt.sparql.values():
+                        add_sparql(g, node, s)
                 elif typename in model.datatypes:
                     dt = model.datatypes[typename]
                     if "pattern" in dt.format: