SPARQL Extensible Value Testing mechanism for registering SPARQL functions from SHACL Advanced Features

[usalu]

The whole advanced SHACL seems quite powerfull. One nice feature is SHACL Functions which requires a mechanism where the shacl shape graph can automatically register SPARQLFunctions to the sparql engine over SPARQL Extensible Value Testing.

When trying to find related issues or source code, I couldn't find anything besides the namespace which obviously mentions the term.

The naive approach that comes to my mind is to use metaprogramming to create a python function that receives a shacl graph as an input and creates a function with the right signature and in the body executes the sparql code with initBindings, and then adds it to CUSTOM_EVALS.

An example shapes graph:

# example.ttl
@prefix ex: <http://example.com/ns#> .
@prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .
@prefix sh: <http://www.w3.org/ns/shacl#> .
@prefix xsd: <http://www.w3.org/2001/XMLSchema#> .
@prefix ex: <http://example.com/ns#> .

ex:multiply
	a sh:SPARQLFunction ;
	rdfs:comment "Multiplies its two arguments $op1 and $op2." ;
	sh:parameter [
		sh:path ex:op1 ;
		sh:datatype xsd:integer ;
		sh:description "The first operand" ;
	] ;
	sh:parameter [
		sh:path ex:op2 ;
		sh:datatype xsd:integer ;
		sh:description "The second operand" ;
	] ;
	sh:returnType xsd:integer ;
	sh:select """
		SELECT ($op1 * $op2 AS ?result)
		WHERE {
		}
		""" .

A first sketch could look something like that.

from rdflib import Graph, Namespace, Literal
from rdflib.namespace import RDF, SH
from rdflib.plugins.sparql import CUSTOM_EVALS
import ast
import copy

def convertExpr2Expression(Expr):
        Expr.lineno = 0
        Expr.col_offset = 0
        result = ast.Expression(Expr.value, lineno=0, col_offset = 0)
        return result

def exec_with_return(code):
    """https://stackoverflow.com/questions/33409207/how-to-return-value-from-exec-in-function"""
    code_ast = ast.parse(code)
    init_ast = copy.deepcopy(code_ast)
    init_ast.body = code_ast.body[:-1]
    last_ast = copy.deepcopy(code_ast)
    last_ast.body = code_ast.body[-1:]
    exec(compile(init_ast, "<ast>", "exec"), globals())
    if type(last_ast.body[0]) == ast.Expr:
        return eval(compile(convertExpr2Expression(last_ast.body[0]), "<ast>", "eval"),globals())
    else:
        exec(compile(last_ast, "<ast>", "exec"),globals())

def registerFunctionsFromShacl(shaclGraph:Graph):
    # TODO: Loop over all functions and subgraphs
    pass  

def registerSparqlFunctionFromShacl(shaclGraph:Graph):
    """Register a SPARQL function from a SHACL shapes graph.
    Assumes the graph has only one SPARQLFunction."""
    uri = shaclGraph.value(None,RDF.type,SH.SPARQLFunction)
    prefix = {n:p for p, n in shaclGraph.namespace_manager.namespaces()}[uri.defrag()]
    name = function_uri.fragment
    curie = prefix + ":" + + name
    function_name = "customShaclFunction_" + prefix + "_" + name
    select = function_uri = shaclGraph.value(uri,SH.select)
    parameters = [ shaclGraph.value(p[2],SH.path).fragment for p in 
                    shaclGraph.triples((uri,SH.parameter,None))]
    for parameter in parameters:
        select = select.replace("$"+parameter,"?"+parameter)
    # Replace PARAMETER with the correct "ctx" or "part" call
    initBindings = "{" + ",".join([ "'" + p + "':" + "PARAMETER" for p in parameters]) + "}"
    # Defining function at runtime and returning address
    # Security issue if shacl graph isn't properly validated beforehand!
    function = exec_with_return(
    (
    f"def {function_name}(ctx, part):"
    f"  if part.name == \"BGP\":"
    f"    triples = []"
    f"    q = {select}"
    f"    ctx.graph.query(q, initBindings={initBindings}))"
    f"    return evalBGP(ctx, triples)"
    f"    raise NotImplementedError()"
    f"{function_name}"
    )
    )
    CUSTOM_EVALS[curie] = function

EX = Namespace("http://example.com/ns#")
g = Graph()
g.add((EX.b1,EX.width,Literal(5)))
g.add((EX.b1,EX.height,Literal(3)))
q = """
SELECT ?subject ?area
WHERE {
    ?subject ex:width ?width .
    ?subject ex:height ?height .
    BIND (ex:multiply(?width, ?height) AS ?area) .
}"""
try:
    g.query(q)
except:
     print ("Not yet registered...")
registerSparqlFunctionFromShacl(Graph().parse("example.ttl"))
print (g.query(q))

As I am not familiar with the internals of rdflib and how

def customEval(ctx, part)

exactly works, I don't know how to completly implement it but it shouldn't be too much work, to get this working tho for someone who is familiar with the logic of it.

Is there interst in such a feature?