Source code for rdflib.plugins.sparql.results.rdfresults

from rdflib import Graph, Namespace, RDF, Variable

from rdflib.query import Result, ResultParser

RS = Namespace("http://www.w3.org/2001/sw/DataAccess/tests/result-set#")


[docs]class RDFResultParser(ResultParser):
[docs] def parse(self, source, **kwargs): return RDFResult(source, **kwargs)
[docs]class RDFResult(Result):
[docs] def __init__(self, source, **kwargs): if not isinstance(source, Graph): graph = Graph() graph.parse(source, **kwargs) else: graph = source rs = graph.value(predicate=RDF.type, object=RS.ResultSet) # there better be only one :) if rs is None: type_ = "CONSTRUCT" # use a new graph g = Graph() g += graph else: askAnswer = graph.value(rs, RS.boolean) if askAnswer is not None: type_ = "ASK" else: type_ = "SELECT" Result.__init__(self, type_) if type_ == "SELECT": self.vars = [Variable(v) for v in graph.objects(rs, RS.resultVariable)] self.bindings = [] for s in graph.objects(rs, RS.solution): sol = {} for b in graph.objects(s, RS.binding): sol[Variable(graph.value(b, RS.variable))] = graph.value( b, RS.value ) self.bindings.append(sol) elif type_ == "ASK": self.askAnswer = askAnswer.value if askAnswer.value is None: raise Exception("Malformed boolean in ask answer!") elif type_ == "CONSTRUCT": self.graph = g