Coverage for rdflib_ocdm / ocdm_graph.py: 86%
198 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-30 21:23 +0000
1#!/usr/bin/python
3# SPDX-FileCopyrightText: 2023-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
4#
5# SPDX-License-Identifier: ISC
7from __future__ import annotations
9from typing import TYPE_CHECKING
11import rdflib
12import rdflib.plugin as plugin
13from rdflib.exceptions import ParserError
14from rdflib.parser import InputSource, Parser, create_input_source
16if TYPE_CHECKING:
17 from typing import IO, BinaryIO, Optional, TextIO, Tuple, Union
19 from rdflib.term import Node as _Node
21 _TripleType = Tuple[_Node, _Node, _Node]
23import pathlib
24import warnings
25from copy import deepcopy
26from datetime import datetime, timedelta, timezone
28from rdflib import Dataset, Graph, URIRef
29from rdflib.term import Node
31from rdflib_ocdm.counter_handler.counter_handler import CounterHandler
32from rdflib_ocdm.graph_utils import _extract_graph_iri, _extract_graph_iri_from_context
33from rdflib_ocdm.prov.provenance import OCDMProvenance
34from rdflib_ocdm.prov.snapshot_entity import SnapshotEntity
37class OCDMGraphCommons:
38 preexisting_graph: Graph | Dataset
40 def __init__(self, counter_handler: CounterHandler):
41 self.__merge_index: dict = dict()
42 self.__entity_index: dict = dict()
43 self.all_entities: set = set()
44 self.provenance = OCDMProvenance(self, counter_handler)
46 def preexisting_finished(
47 self,
48 resp_agent: str | None = None,
49 primary_source: str | None = None,
50 c_time: float | str | None = None,
51 ) -> None:
52 assert isinstance(self, (Graph, Dataset))
53 self.preexisting_graph = deepcopy(self)
55 unique_subjects: set = set()
56 if isinstance(self, Dataset):
57 for s, _, _, _ in self.quads((None, None, None, None)):
58 unique_subjects.add(s)
59 else:
60 unique_subjects = set(self.subjects(unique=True))
62 for subject in unique_subjects:
63 existing_graph_iri = self.entity_index.get(subject, {}).get("graph_iri")
64 self.entity_index[subject] = {
65 "to_be_deleted": False,
66 "is_restored": False,
67 "resp_agent": resp_agent,
68 "source": primary_source,
69 "graph_iri": existing_graph_iri,
70 }
72 if isinstance(self, Dataset) and existing_graph_iri is None: 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true
73 self.entity_index[subject]["graph_iri"] = _extract_graph_iri(
74 self, subject
75 )
77 self.all_entities.add(subject)
78 count = self.provenance.counter_handler.read_counter(str(subject))
79 if count == 0:
80 if c_time is None:
81 cur_time = (
82 datetime.now(tz=timezone.utc).replace(microsecond=0)
83 - timedelta(seconds=5)
84 ).isoformat(sep="T")
85 else:
86 cur_time = (
87 datetime.fromtimestamp(float(c_time), tz=timezone.utc).replace(
88 microsecond=0
89 )
90 - timedelta(seconds=5)
91 ).isoformat(sep="T")
92 new_snapshot: SnapshotEntity = self.provenance._create_snapshot(
93 URIRef(str(subject)), cur_time
94 )
95 new_snapshot.has_description(
96 f"The entity '{str(subject)}' has been created."
97 )
99 def merge(self, res: URIRef, other: URIRef) -> None:
100 assert isinstance(self, (Graph, Dataset))
101 other_graph_iri = None
102 if isinstance(self, Dataset):
103 other_graph_iri = _extract_graph_iri(self, other)
105 quads_list = list(self.quads((None, None, other, None)))
106 for s, p, o, c in quads_list:
107 self.remove((s, p, o, c)) # type: ignore[arg-type]
108 self.add((s, p, res, c)) # type: ignore[arg-type]
109 quads_list_del = list(self.quads((other, None, None, None)))
110 for s, p, o, c in quads_list_del:
111 self.remove((s, p, o, c)) # type: ignore[arg-type]
112 elif isinstance(self, Graph): 112 ↛ 122line 112 didn't jump to line 122 because the condition on line 112 was always true
113 triples_list = list(self.triples((None, None, other)))
114 for triple in triples_list:
115 self.remove(triple)
116 new_triple = (triple[0], triple[1], res)
117 self.add(new_triple)
118 triples_list_del = list(self.triples((other, None, None)))
119 for triple in triples_list_del:
120 self.remove(triple)
122 self._OCDMGraphCommons__merge_index.setdefault(res, set()).add(other)
123 if other not in self.entity_index:
124 self.entity_index[other] = {
125 "to_be_deleted": False,
126 "is_restored": False,
127 "resp_agent": None,
128 "source": None,
129 "graph_iri": other_graph_iri,
130 }
131 else:
132 if ( 132 ↛ 136line 132 didn't jump to line 136 because the condition on line 132 was never true
133 other_graph_iri is not None
134 and self.entity_index[other].get("graph_iri") is None
135 ):
136 self.entity_index[other]["graph_iri"] = other_graph_iri
137 self.entity_index[other]["to_be_deleted"] = True
139 def mark_as_deleted(self, res: URIRef) -> None:
140 self.entity_index[res]["to_be_deleted"] = True
142 def mark_as_restored(self, res: URIRef) -> None:
143 """
144 Marks an entity as being restored after deletion.
145 This will:
146 1. Set is_restored flag to True in the entity_index
147 2. Set to_be_deleted flag to False
149 :param res: The URI reference of the entity to restore
150 :type res: URIRef
151 :return: None
152 """
153 if res in self.entity_index: 153 ↛ exitline 153 didn't return from function 'mark_as_restored' because the condition on line 153 was always true
154 self.entity_index[res]["is_restored"] = True
155 self.entity_index[res]["to_be_deleted"] = False
157 @property
158 def merge_index(self) -> dict:
159 return self.__merge_index
161 @property
162 def entity_index(self) -> dict:
163 return self.__entity_index
165 def generate_provenance(self, c_time: float | None = None) -> None:
166 return self.provenance.generate_provenance(c_time)
168 def get_entity(self, res: str) -> SnapshotEntity | None:
169 entity = self.provenance.get_entity(res)
170 if isinstance(entity, SnapshotEntity):
171 return entity
172 return None
174 def commit_changes(self) -> None:
175 self._OCDMGraphCommons__merge_index = dict()
176 self._OCDMGraphCommons__entity_index = dict()
177 assert isinstance(self, (Graph, Dataset))
178 self.preexisting_graph = deepcopy(self)
180 def get_provenance_graphs(self) -> Dataset:
181 prov_g = Dataset()
182 for _, prov_entity in self.provenance.res_to_entity.items():
183 for triple in prov_entity.g.triples((None, None, None)):
184 prov_iri = URIRef(prov_entity.prov_subject + "/prov/")
185 prov_g.add((triple[0], triple[1], triple[2], prov_iri)) # type: ignore[arg-type]
186 return prov_g
189class OCDMGraph(OCDMGraphCommons, Graph):
190 def __init__(self, counter_handler: CounterHandler | None = None):
191 Graph.__init__(self)
192 self.preexisting_graph: Graph | Dataset = Graph()
193 OCDMGraphCommons.__init__(self, counter_handler) # type: ignore[arg-type]
195 def add(
196 self,
197 triple: _TripleType,
198 resp_agent: object = None,
199 primary_source: object = None,
200 ): # type: ignore[override]
201 s, p, o = triple
202 assert isinstance(s, Node), "Subject %s must be an rdflib term" % (s,)
203 assert isinstance(p, Node), "Predicate %s must be an rdflib term" % (p,)
204 assert isinstance(o, Node), "Object %s must be an rdflib term" % (o,)
205 self.store.add((s, p, o), self, quoted=False)
207 # Add the subject to all_entities if it's not already present
208 if s not in self.all_entities:
209 self.all_entities.add(s)
211 if s not in self.entity_index:
212 self.entity_index[s] = {
213 "to_be_deleted": False,
214 "is_restored": False,
215 "resp_agent": resp_agent,
216 "source": primary_source,
217 }
219 return self
221 def parse(
222 self,
223 source: Optional[
224 Union[IO[bytes], TextIO, InputSource, str, bytes, pathlib.PurePath]
225 ] = None,
226 publicID: Optional[str] = None, # noqa: N803
227 format: Optional[str] = None,
228 location: Optional[str] = None,
229 file: Optional[Union[BinaryIO, TextIO]] = None,
230 data: Optional[Union[str, bytes]] = None,
231 resp_agent: URIRef | None = None,
232 primary_source: URIRef | None = None,
233 **args: object,
234 ) -> Graph:
235 source = create_input_source(
236 source=source,
237 publicID=publicID,
238 location=location,
239 file=file,
240 data=data,
241 format=format,
242 )
243 if format is None: 243 ↛ 245line 243 didn't jump to line 245 because the condition on line 243 was always true
244 format = source.content_type
245 could_not_guess_format = False
246 if format is None: 246 ↛ 257line 246 didn't jump to line 257 because the condition on line 246 was always true
247 _file = getattr(source, "file", None)
248 if ( 248 ↛ 254line 248 didn't jump to line 254 because the condition on line 248 was always true
249 _file is not None
250 and getattr(_file, "name", None)
251 and isinstance(_file.name, str)
252 ):
253 format = rdflib.util.guess_format(_file.name)
254 if format is None: 254 ↛ 255line 254 didn't jump to line 255 because the condition on line 254 was never true
255 format = "turtle"
256 could_not_guess_format = True
257 parser = plugin.get(format, Parser)()
258 try:
259 parser.parse(source, self, **args)
260 except SyntaxError as se:
261 if could_not_guess_format:
262 raise ParserError(
263 "Could not guess RDF format for %r from file"
264 " extension so tried Turtle but failed."
265 " You can explicitly specify format using"
266 " the format argument." % source
267 )
268 else:
269 raise se
270 finally:
271 if source.auto_close: 271 ↛ 274line 271 didn't jump to line 274 because the condition on line 271 was always true
272 source.close()
274 for subject in self.subjects(unique=True):
275 if subject not in self.all_entities: 275 ↛ 276line 275 didn't jump to line 276 because the condition on line 275 was never true
276 self.all_entities.add(subject)
278 if subject not in self.entity_index: 278 ↛ 279line 278 didn't jump to line 279 because the condition on line 278 was never true
279 self.entity_index[subject] = {
280 "to_be_deleted": False,
281 "is_restored": False,
282 "resp_agent": resp_agent,
283 "source": primary_source,
284 }
286 return self
289class OCDMDataset(OCDMGraphCommons, Dataset):
290 def __init__(self, counter_handler: CounterHandler | None = None):
291 Dataset.__init__(self)
292 self.preexisting_graph: Graph | Dataset = Dataset()
293 OCDMGraphCommons.__init__(self, counter_handler) # type: ignore[arg-type]
295 def __deepcopy__(self, memo):
296 new_graph = OCDMDataset(counter_handler=self.provenance.counter_handler)
298 # Copy graph data
299 for quad in self.quads((None, None, None, None)):
300 new_graph.add(quad) # type: ignore[arg-type]
302 # Copy entity index and metadata
303 for key, value in self.entity_index.items():
304 new_graph.entity_index[key] = value.copy()
305 new_graph.all_entities = self.all_entities.copy()
306 for key, value in self.merge_index.items(): 306 ↛ 307line 306 didn't jump to line 307 because the loop on line 306 never started
307 new_graph._OCDMGraphCommons__merge_index[key] = value.copy() # type: ignore[attr-defined]
309 return new_graph
311 def add( # type: ignore[override]
312 self,
313 triple_or_quad: tuple[Node, Node, Node] | tuple[Node, Node, Node, Graph | None],
314 resp_agent: object = None,
315 primary_source: object = None,
316 ) -> Dataset:
318 s, p, o, c = self._spoc(triple_or_quad, default=True)
320 _assertnode(s, p, o)
322 self.store.add(
323 (s, p, o),
324 context=c, # type: ignore[arg-type]
325 quoted=False,
326 )
328 # Add the subject to all_entities if it's not already present
329 if s not in self.all_entities:
330 self.all_entities.add(s)
332 if s not in self.entity_index:
333 self.entity_index[s] = {
334 "to_be_deleted": False,
335 "is_restored": False,
336 "resp_agent": resp_agent,
337 "source": primary_source,
338 "graph_iri": None,
339 }
341 # Store graph_iri in entity_index for later retrieval
342 # We already have the context from _spoc, use it directly for efficiency
343 if self.entity_index[s]["graph_iri"] is None:
344 self.entity_index[s]["graph_iri"] = _extract_graph_iri_from_context(c)
346 return self
348 def parse( # type: ignore[override]
349 self,
350 source: IO[bytes]
351 | TextIO
352 | InputSource
353 | str
354 | bytes
355 | pathlib.PurePath
356 | None = None,
357 publicID: str | None = None, # noqa: N803
358 format: str | None = None,
359 location: str | None = None,
360 file: BinaryIO | TextIO | None = None,
361 data: str | bytes | None = None,
362 resp_agent: URIRef | None = None,
363 primary_source: URIRef | None = None,
364 **args: object,
365 ) -> Graph:
366 source = create_input_source(
367 source=source,
368 publicID=publicID,
369 location=location,
370 file=file,
371 data=data,
372 format=format,
373 )
375 g_id = publicID or source.getPublicId() or ""
376 if not isinstance(g_id, Node): 376 ↛ 377line 376 didn't jump to line 377 because the condition on line 376 was never true
377 g_id = URIRef(g_id)
379 context = Graph(store=self.store, identifier=g_id)
380 context.remove((None, None, None)) # type: ignore[arg-type]
381 context.parse(source, publicID=publicID, format=format, **args) # type: ignore[arg-type]
382 # TODO: FIXME: This should not return context, but self.
384 unique_subjects = set()
385 for s, _, _, _ in self.quads((None, None, None, None)):
386 unique_subjects.add(s)
388 for subject in unique_subjects:
389 if subject not in self.all_entities: 389 ↛ 392line 389 didn't jump to line 392 because the condition on line 389 was always true
390 self.all_entities.add(subject)
392 if subject not in self.entity_index: 392 ↛ 402line 392 didn't jump to line 402 because the condition on line 392 was always true
393 self.entity_index[subject] = {
394 "to_be_deleted": False,
395 "is_restored": False,
396 "resp_agent": resp_agent,
397 "source": primary_source,
398 "graph_iri": None,
399 }
401 # Store graph_iri for this subject by finding its context
402 if ( 402 ↛ 388line 402 didn't jump to line 388 because the condition on line 402 was always true
403 "graph_iri" not in self.entity_index[subject]
404 or self.entity_index[subject]["graph_iri"] is None
405 ):
406 self.entity_index[subject]["graph_iri"] = _extract_graph_iri(
407 self, subject
408 )
410 return context
413def _assertnode(*terms):
414 for t in terms:
415 assert isinstance(t, Node), "Term %s must be an rdflib term" % (t,)
416 return True
419# Backward compatibility alias
420class OCDMConjunctiveGraph(OCDMDataset):
421 """
422 Deprecated: Use OCDMDataset instead.
424 This class is maintained for backward compatibility only.
425 OCDMConjunctiveGraph has been renamed to OCDMDataset to reflect
426 the migration from the deprecated ConjunctiveGraph to Dataset.
427 """
429 def __init__(self, counter_handler: CounterHandler | None = None):
430 warnings.warn(
431 "OCDMConjunctiveGraph is deprecated, use OCDMDataset instead",
432 DeprecationWarning,
433 stacklevel=2,
434 )
435 super().__init__(counter_handler)