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.load(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