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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
|
"""
Code for tying SPARQL Engine into RDFLib
These should be automatically registered with RDFLib
"""
from __future__ import annotations
from typing import Any, Mapping, Optional, Union
from rdflib.graph import Graph
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, Update
from rdflib.plugins.sparql.update import evalUpdate
from rdflib.query import Processor, Result, UpdateProcessor
from rdflib.term import Identifier
def prepareQuery(
queryString: str, initNs: Mapping[str, Any] = {}, base: Optional[str] = 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: str, initNs: Mapping[str, Any] = {}, base: Optional[str] = None
) -> Update:
"""
Parse and translate a SPARQL Update
"""
ret = translateUpdate(parseUpdate(updateString), base, initNs)
ret._original_args = (updateString, initNs, base)
return ret
def processUpdate(
graph: Graph,
updateString: str,
initBindings: Mapping[str, Identifier] = {},
initNs: Mapping[str, Any] = {},
base: Optional[str] = None,
) -> 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: Mapping[str, Any]):
Result.__init__(self, res["type_"])
self.vars = res.get("vars_")
# type error: Incompatible types in assignment (expression has type "Optional[Any]", variable has type "MutableSequence[Mapping[Variable, Identifier]]")
self.bindings = res.get("bindings") # type: ignore[assignment]
self.askAnswer = res.get("askAnswer")
self.graph = res.get("graph")
class SPARQLUpdateProcessor(UpdateProcessor):
def __init__(self, graph):
self.graph = graph
def update(
self,
strOrQuery: Union[str, Update],
initBindings: Mapping[str, Identifier] = {},
initNs: Mapping[str, Any] = {},
) -> None:
"""
.. caution::
This method can access indirectly requested network endpoints, for
example, query processing will attempt to access network endpoints
specified in ``SERVICE`` directives.
When processing untrusted or potentially malicious queries, measures
should be taken to restrict network and file access.
For information on available security measures, see the RDFLib
:doc:`Security Considerations </security_considerations>`
documentation.
"""
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
# NOTE on type error: this is because the super type constructor does not
# accept base argument and thie position of the DEBUG argument is
# different.
# type error: Signature of "query" incompatible with supertype "Processor"
def query( # type: ignore[override]
self,
strOrQuery: Union[str, Query],
initBindings: Mapping[str, Identifier] = {},
initNs: Mapping[str, Any] = {},
base: Optional[str] = None,
DEBUG: bool = False,
) -> Mapping[str, Any]:
"""
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.
.. caution::
This method can access indirectly requested network endpoints, for
example, query processing will attempt to access network endpoints
specified in ``SERVICE`` directives.
When processing untrusted or potentially malicious queries, measures
should be taken to restrict network and file access.
For information on available security measures, see the RDFLib
:doc:`Security Considerations </security_considerations>`
documentation.
"""
if not isinstance(strOrQuery, Query):
parsetree = parseQuery(strOrQuery)
query = translateQuery(parsetree, base, initNs)
else:
query = strOrQuery
return evalQuery(self.graph, query, initBindings, base)
|
