-
Notifications
You must be signed in to change notification settings - Fork 544
/
processor.py
81 lines (61 loc) · 2.5 KB
/
processor.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
"""
Code for tying SPARQL Engine into RDFLib
These should be automatically registered with RDFLib
"""
from rdflib.plugins.sparql.algebra import translateQuery, translateUpdate
from rdflib.plugins.sparql.evaluate import evalQuery
from rdflib.plugins.sparql.parser import parseQuery, parseUpdate
from rdflib.plugins.sparql.sparql import Query
from rdflib.plugins.sparql.update import evalUpdate
from rdflib.query import Processor, Result, UpdateProcessor
def prepareQuery(queryString, initNs={}, base=None) -> Query:
"""
Parse and translate a SPARQL Query
"""
ret = translateQuery(parseQuery(queryString), base, initNs)
ret._original_args = (queryString, initNs, base)
return ret
def prepareUpdate(updateString, initNs={}, base=None):
"""
Parse and translate a SPARQL Update
"""
ret = translateUpdate(parseUpdate(updateString), base, initNs)
ret._original_args = (updateString, initNs, base)
return ret
def processUpdate(graph, updateString, initBindings={}, initNs={}, base=None):
"""
Process a SPARQL Update Request
returns Nothing on success or raises Exceptions on error
"""
evalUpdate(
graph, translateUpdate(parseUpdate(updateString), base, initNs), initBindings
)
class SPARQLResult(Result):
def __init__(self, res):
Result.__init__(self, res["type_"])
self.vars = res.get("vars_")
self.bindings = res.get("bindings")
self.askAnswer = res.get("askAnswer")
self.graph = res.get("graph")
class SPARQLUpdateProcessor(UpdateProcessor):
def __init__(self, graph):
self.graph = graph
def update(self, strOrQuery, initBindings={}, initNs={}):
if isinstance(strOrQuery, str):
strOrQuery = translateUpdate(parseUpdate(strOrQuery), initNs=initNs)
return evalUpdate(self.graph, strOrQuery, initBindings)
class SPARQLProcessor(Processor):
def __init__(self, graph):
self.graph = graph
def query(self, strOrQuery, initBindings={}, initNs={}, base=None, DEBUG=False):
"""
Evaluate a query with the given initial bindings, and initial
namespaces. The given base is used to resolve relative URIs in
the query and will be overridden by any BASE given in the query.
"""
if not isinstance(strOrQuery, Query):
parsetree = parseQuery(strOrQuery)
query = translateQuery(parsetree, base, initNs)
else:
query = strOrQuery
return evalQuery(self.graph, query, initBindings, base)