Coverage for target/native-coverage/profiles/8cb89e7529e5488ca16399c0daa3ee58/project/.venv/lib/python3.13/site-packages/rdflib_turbo/graph.py: 100%
215 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-20 21:00 +0000
1"""RDFLib's Python facade over one exact-term native Store or transaction.
3RDFLib supplies terms, namespace management and parser/serializer boundaries.
4Its in-memory graph stores and Python SPARQL evaluators never execute here.
5"""
7import json
8from contextlib import contextmanager
10from rdflib import BNode, URIRef, Variable
11from rdflib import Dataset as RDFLibDataset
12from rdflib import Graph as RDFLibGraph
13from rdflib.graph import DATASET_DEFAULT_GRAPH_ID
14from rdflib.parser import create_input_source
15from rdflib.query import Result
16from rdflib.store import Store as ReferenceStore
17from rdflib.util import from_n3, guess_format
19from ._namespaces import NamespaceBindings
20from ._operations import _encode, _term
21from ._reads import ProjectedReads, encode, encode_graph, scoped_rows
22from .store import Store, Transaction
25class RDFLibStore(ReferenceStore):
26 """Context-aware RDFLib Store contract backed exclusively by native storage."""
28 context_aware = True
29 graph_aware = True
30 transaction_aware = True
31 formula_aware = False
33 def __init__(
34 self, backend=None, *, default_identifier=DATASET_DEFAULT_GRAPH_ID, namespaces=None
35 ):
36 super().__init__()
37 if backend is not None and not isinstance(backend, (Store, Transaction)):
38 raise TypeError("Native Store or Transaction required")
39 self.backend = Store() if backend is None else backend
40 self.default_identifier = default_identifier
41 if namespaces is None:
42 if not hasattr(self.backend, "_rdflib_namespaces"):
43 self.backend._rdflib_namespaces = NamespaceBindings()
44 namespaces = self.backend._rdflib_namespaces
45 else:
46 namespaces = NamespaceBindings(namespaces)
47 self.backend._rdflib_namespaces = namespaces
48 self._namespaces = namespaces
50 def _graph(self, context):
51 if context is None:
52 return None
53 identifier = context.identifier if hasattr(context, "identifier") else context
54 return "" if identifier == self.default_identifier else json.dumps(_encode(identifier))
56 def _pattern(self, triple, context):
57 return json.dumps([*map(_encode, triple), self._graph(context)])
59 def add(self, triple, context, quoted=False):
60 if quoted:
61 raise ValueError("Quoted graphs are unsupported")
62 if context is None:
63 raise ValueError("Graph context is required")
64 self.backend._native.insert_terms(*map(encode, triple), encode_graph(self, context))
66 def remove(self, triple, context=None):
67 self.backend._native.remove_pattern(self._pattern(triple, context))
69 def triples(self, triple_pattern, context=None):
70 if context is not None and isinstance(self.backend, Store):
71 for triple in scoped_rows(self, triple_pattern, context):
72 yield triple, iter((context,))
73 return
74 rows = json.loads(self.backend._native.match_quads(self._pattern(triple_pattern, context)))
75 grouped = {}
76 for s, p, o, g in rows:
77 triple = (_term(s), _term(p), _term(o))
78 identifier = self.default_identifier if g is None else _term(g)
79 grouped.setdefault(triple, []).append(identifier)
80 for triple, contexts in grouped.items():
81 yield (
82 triple,
83 (
84 Graph(store=self, identifier=identifier, bind_namespaces="none")
85 for identifier in contexts
86 ),
87 )
89 def __len__(self, context=None):
90 if context is None:
91 return self.backend._native.count_pattern("[null,null,null,null]")
92 return self.backend._native.graph_len(encode_graph(self, context))
94 def contexts(self, triple=None):
95 if triple is not None:
96 identifiers = {context for _, contexts in self.triples(triple) for context in contexts}
97 else:
98 identifiers = {
99 from_n3(value) for value in json.loads(self.backend._native.named_graphs())
100 }
101 if len(self.default_graph()):
102 identifiers.add(self.default_identifier)
103 yield from identifiers
105 def default_graph(self):
106 return Graph(store=self, identifier=self.default_identifier)
108 def add_graph(self, graph):
109 if (
110 graph.identifier != self.default_identifier
111 and not self.backend._native.contains_named_graph(encode(graph.identifier))
112 ):
113 self.backend._native.change_graph(self._graph(graph), False)
115 def remove_graph(self, graph):
116 name = self._graph(graph)
117 if name:
118 self.backend._native.change_graph(name, True)
119 else:
120 self.remove((None, None, None), graph)
122 def bind(self, prefix, namespace, override=True):
123 self._namespaces.bind(prefix, namespace, override)
125 def prefix(self, namespace):
126 return self._namespaces.prefix(namespace)
128 def namespace(self, prefix):
129 return self._namespaces.namespace(prefix)
131 def namespaces(self):
132 yield from self._namespaces.items()
134 def _query_source(self, source, initNs, initBindings, kwargs):
135 if not isinstance(source, str):
136 raise TypeError("Native SPARQL currently requires a query string")
137 if initBindings:
138 raise NotImplementedError("Native initial query bindings are not implemented")
139 if kwargs:
140 raise TypeError(f"Unsupported native query options: {sorted(kwargs)}")
141 return (
142 "\n".join(f"PREFIX {prefix}: <{uri}>" for prefix, uri in initNs.items()) + "\n" + source
143 )
145 def query(self, query, initNs, initBindings, queryGraph, **kwargs):
146 source = self._query_source(query, initNs, initBindings, kwargs)
147 if queryGraph == "__UNION__":
148 kind, payload = self.backend._native.query(source, True)
149 else:
150 graph = "" if queryGraph is None else self._graph(queryGraph)
151 kind, payload = self.backend._native.query_graph(source, graph)
152 result = Result(kind)
153 if kind == "ASK":
154 result.askAnswer = json.loads(payload)
155 elif kind == "SELECT":
156 value = json.loads(payload)
157 result.vars = [Variable(name) for name in value["variables"]]
158 result.bindings = [
159 {Variable(name): _term(term) for name, term in row.items()} for row in value["rows"]
160 ]
161 else:
162 # A requested graph result is newly owned output, never a copy of input storage.
163 result.graph = Graph().parse(data=payload, format="nt")
164 return result
166 def update(self, update, initNs, initBindings, queryGraph, **kwargs):
167 source = self._query_source(update, initNs, initBindings, kwargs)
168 if queryGraph not in (None, "__UNION__", self.default_identifier):
169 raise NotImplementedError("Scoped named-graph SPARQL Update is not implemented")
170 self.backend.update(source)
172 def commit(self):
173 raise RuntimeError("Use Graph.transaction() or StoreDataset.transaction() scoped context")
175 def rollback(self):
176 raise RuntimeError("Use Graph.transaction() or StoreDataset.transaction() scoped context")
178 def close(self, commit_pending_transaction=False):
179 if commit_pending_transaction:
180 raise ValueError("Close cannot commit a scoped transaction")
181 if isinstance(self.backend, Transaction):
182 raise RuntimeError("Close the enclosing transaction context")
183 self.backend.close()
186class Graph(ProjectedReads, RDFLibGraph):
187 """RDFLib Graph facade with exact-term native storage and native SPARQL."""
189 _turbo_native_graph = True
191 def __init__(
192 self,
193 store="default",
194 identifier=None,
195 namespace_manager=None,
196 base=None,
197 bind_namespaces="rdflib",
198 ):
199 identifier = (
200 BNode()
201 if identifier is None
202 else identifier
203 if isinstance(identifier, (URIRef, BNode))
204 else URIRef(identifier)
205 )
206 if store == "default":
207 store = RDFLibStore(default_identifier=identifier)
208 elif isinstance(store, (Store, Transaction)):
209 store = RDFLibStore(store)
210 elif not isinstance(store, RDFLibStore):
211 raise TypeError("Graph requires a native RDFLibStore, Store or Transaction")
212 super().__init__(
213 store=store,
214 identifier=identifier,
215 namespace_manager=namespace_manager,
216 base=base,
217 bind_namespaces=bind_namespaces,
218 )
220 def query(
221 self,
222 query_object,
223 processor="sparql",
224 result="sparql",
225 initNs=None,
226 initBindings=None,
227 use_store_provided=True,
228 **kwargs,
229 ):
230 if processor != "sparql" or result != "sparql" or not use_store_provided:
231 raise ValueError("Graph only supports its native SPARQL processor")
232 return self.store.query(
233 query_object,
234 dict(self.namespaces()) if initNs is None else initNs,
235 {} if initBindings is None else initBindings,
236 self.identifier,
237 **kwargs,
238 )
240 def update(
241 self,
242 update_object,
243 processor="sparql",
244 initNs=None,
245 initBindings=None,
246 use_store_provided=True,
247 **kwargs,
248 ):
249 if processor != "sparql" or not use_store_provided:
250 raise ValueError("Graph only supports its native SPARQL processor")
251 return self.store.update(
252 update_object,
253 dict(self.namespaces()) if initNs is None else initNs,
254 {} if initBindings is None else initBindings,
255 self.identifier,
256 **kwargs,
257 )
259 def parse(
260 self, source=None, publicID=None, format=None, location=None, file=None, data=None, **args
261 ):
262 if args:
263 raise TypeError(f"Unsupported native parser options: {sorted(args)}")
264 if isinstance(self.store.backend, Transaction):
265 try:
266 input_source = create_input_source(
267 source=source, publicID=publicID, location=location, file=file, data=data
268 )
269 try:
270 stream = input_source.getCharacterStream() or input_source.getByteStream()
271 document = stream.read()
272 if isinstance(document, bytes):
273 document = document.decode("utf-8")
274 rdf_format = (
275 format
276 or input_source.content_type
277 or guess_format(input_source.getPublicId())
278 or "turtle"
279 )
280 prefixes = json.loads(
281 self.store.backend._native.load_graph(
282 document,
283 rdf_format,
284 publicID or input_source.getPublicId() or self.base,
285 "" if isinstance(self, StoreDataset) else self.store._graph(self),
286 isinstance(self, StoreDataset),
287 )
288 )
289 for prefix, namespace in prefixes:
290 self.bind(prefix, URIRef(namespace))
291 finally:
292 if input_source.auto_close:
293 input_source.close()
294 except BaseException:
295 self.store.backend._native.rollback()
296 raise
297 else:
298 with self.transaction() as graph:
299 graph.parse(source, publicID, format, location, file, data)
300 return self
302 @contextmanager
303 def transaction(self):
304 if isinstance(self.store.backend, Transaction):
305 raise RuntimeError("Nested transactions are unsupported")
306 namespaces = self.store._namespaces.copy()
307 with self.store.backend.transaction() as transaction:
308 store = RDFLibStore(
309 transaction, default_identifier=self.store.default_identifier, namespaces=namespaces
310 )
311 yield Graph(
312 store=store, identifier=self.identifier, base=self.base, bind_namespaces="none"
313 )
314 self.store._namespaces.replace(store._namespaces)
317class StoreDataset(RDFLibDataset):
318 """RDFLib Dataset foundation over the transactional Store.
320 Inference/SHACL remain on the separate legacy Dataset until direct adapters exist.
321 """
323 _turbo_native_graph = True
325 def __init__(self, store="default", default_union=False, default_graph_base=None):
326 if store == "default":
327 store = RDFLibStore()
328 elif isinstance(store, (Store, Transaction)):
329 store = RDFLibStore(store)
330 elif not isinstance(store, RDFLibStore):
331 raise TypeError("StoreDataset requires native storage")
332 super().__init__(
333 store=store, default_union=default_union, default_graph_base=default_graph_base
334 )
335 self.default_graph = Graph(
336 store=self.store, identifier=DATASET_DEFAULT_GRAPH_ID, base=default_graph_base
337 )
339 def graph(self, identifier=None, base=None):
340 graph = Graph(store=self.store, identifier=identifier, base=base)
341 self.store.add_graph(graph)
342 return graph
344 def get_context(self, identifier, quoted=False, base=None):
345 if quoted:
346 raise ValueError("Quoted graphs are unsupported")
347 return Graph(store=self.store, identifier=identifier, base=base)
349 def query(
350 self,
351 query_object,
352 processor="sparql",
353 result="sparql",
354 initNs=None,
355 initBindings=None,
356 use_store_provided=True,
357 **kwargs,
358 ):
359 if processor != "sparql" or result != "sparql" or not use_store_provided:
360 raise ValueError("StoreDataset only supports its native SPARQL processor")
361 return self.store.query(
362 query_object,
363 dict(self.namespaces()) if initNs is None else initNs,
364 {} if initBindings is None else initBindings,
365 "__UNION__" if self.default_union else DATASET_DEFAULT_GRAPH_ID,
366 **kwargs,
367 )
369 def update(
370 self,
371 update_object,
372 processor="sparql",
373 initNs=None,
374 initBindings=None,
375 use_store_provided=True,
376 **kwargs,
377 ):
378 if processor != "sparql" or not use_store_provided:
379 raise ValueError("StoreDataset only supports its native SPARQL processor")
380 return self.store.update(
381 update_object,
382 dict(self.namespaces()) if initNs is None else initNs,
383 {} if initBindings is None else initBindings,
384 DATASET_DEFAULT_GRAPH_ID,
385 **kwargs,
386 )
388 parse = Graph.parse
390 @contextmanager
391 def transaction(self):
392 if isinstance(self.store.backend, Transaction):
393 raise RuntimeError("Nested transactions are unsupported")
394 namespaces = self.store._namespaces.copy()
395 with self.store.backend.transaction() as transaction:
396 store = RDFLibStore(transaction, namespaces=namespaces)
397 yield StoreDataset(store=store, default_union=self.default_union)
398 self.store._namespaces.replace(store._namespaces)