Coverage for rdflib_ocdm/ocdm_graph.py: 86%
199 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-21 14:04 +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 incoming_quads = list(self.quads((None, None, other, None)))
106 for s, p, o, c in incoming_quads:
107 self.remove((s, p, o, c)) # type: ignore[arg-type]
108 self.add((s, p, res, c)) # type: ignore[arg-type]
109 outgoing_quads = list(self.quads((other, None, None, None)))
110 for s, p, o, c in outgoing_quads:
111 self.remove((s, p, o, c)) # type: ignore[arg-type]
112 self.add((res, p, o, c)) # type: ignore[arg-type]
113 elif isinstance(self, Graph): 113 ↛ 123line 113 didn't jump to line 123 because the condition on line 113 was always true
114 incoming_triples = list(self.triples((None, None, other)))
115 for s, p, o in incoming_triples:
116 self.remove((s, p, o))
117 self.add((s, p, res))
118 outgoing_triples = list(self.triples((other, None, None)))
119 for s, p, o in outgoing_triples:
120 self.remove((s, p, o))
121 self.add((res, p, o))
123 self._OCDMGraphCommons__merge_index.setdefault(res, set()).add(other)
124 if other not in self.entity_index:
125 self.entity_index[other] = {
126 "to_be_deleted": False,
127 "is_restored": False,
128 "resp_agent": None,
129 "source": None,
130 "graph_iri": other_graph_iri,
131 }
132 else:
133 if ( 133 ↛ 137line 133 didn't jump to line 137 because the condition on line 133 was never true
134 other_graph_iri is not None
135 and self.entity_index[other].get("graph_iri") is None
136 ):
137 self.entity_index[other]["graph_iri"] = other_graph_iri
138 self.entity_index[other]["to_be_deleted"] = True
140 def mark_as_deleted(self, res: URIRef) -> None:
141 self.entity_index[res]["to_be_deleted"] = True
143 def mark_as_restored(self, res: URIRef) -> None:
144 """
145 Marks an entity as being restored after deletion.
146 This will:
147 1. Set is_restored flag to True in the entity_index
148 2. Set to_be_deleted flag to False
150 :param res: The URI reference of the entity to restore
151 :type res: URIRef
152 :return: None
153 """
154 if res in self.entity_index: 154 ↛ exitline 154 didn't return from function 'mark_as_restored' because the condition on line 154 was always true
155 self.entity_index[res]["is_restored"] = True
156 self.entity_index[res]["to_be_deleted"] = False
158 @property
159 def merge_index(self) -> dict:
160 return self.__merge_index
162 @property
163 def entity_index(self) -> dict:
164 return self.__entity_index
166 def generate_provenance(self, c_time: float | None = None) -> None:
167 return self.provenance.generate_provenance(c_time)
169 def get_entity(self, res: str) -> SnapshotEntity | None:
170 entity = self.provenance.get_entity(res)
171 if isinstance(entity, SnapshotEntity):
172 return entity
173 return None
175 def commit_changes(self) -> None:
176 self._OCDMGraphCommons__merge_index = dict()
177 self._OCDMGraphCommons__entity_index = dict()
178 assert isinstance(self, (Graph, Dataset))
179 self.preexisting_graph = deepcopy(self)
181 def get_provenance_graphs(self) -> Dataset:
182 prov_g = Dataset()
183 for _, prov_entity in self.provenance.res_to_entity.items():
184 for triple in prov_entity.g.triples((None, None, None)):
185 prov_iri = URIRef(prov_entity.prov_subject + "/prov/")
186 prov_g.add((triple[0], triple[1], triple[2], prov_iri)) # type: ignore[arg-type]
187 return prov_g
190class OCDMGraph(OCDMGraphCommons, Graph):
191 def __init__(self, counter_handler: CounterHandler | None = None):
192 Graph.__init__(self)
193 self.preexisting_graph: Graph | Dataset = Graph()
194 OCDMGraphCommons.__init__(self, counter_handler) # type: ignore[arg-type]
196 def add(
197 self,
198 triple: _TripleType,
199 resp_agent: object = None,
200 primary_source: object = None,
201 ): # type: ignore[override]
202 s, p, o = triple
203 assert isinstance(s, Node), "Subject %s must be an rdflib term" % (s,)
204 assert isinstance(p, Node), "Predicate %s must be an rdflib term" % (p,)
205 assert isinstance(o, Node), "Object %s must be an rdflib term" % (o,)
206 self.store.add((s, p, o), self, quoted=False)
208 # Add the subject to all_entities if it's not already present
209 if s not in self.all_entities:
210 self.all_entities.add(s)
212 if s not in self.entity_index:
213 self.entity_index[s] = {
214 "to_be_deleted": False,
215 "is_restored": False,
216 "resp_agent": resp_agent,
217 "source": primary_source,
218 }
220 return self
222 def parse(
223 self,
224 source: Optional[
225 Union[IO[bytes], TextIO, InputSource, str, bytes, pathlib.PurePath]
226 ] = None,
227 publicID: Optional[str] = None, # noqa: N803
228 format: Optional[str] = None,
229 location: Optional[str] = None,
230 file: Optional[Union[BinaryIO, TextIO]] = None,
231 data: Optional[Union[str, bytes]] = None,
232 resp_agent: URIRef | None = None,
233 primary_source: URIRef | None = None,
234 **args: object,
235 ) -> Graph:
236 source = create_input_source(
237 source=source,
238 publicID=publicID,
239 location=location,
240 file=file,
241 data=data,
242 format=format,
243 )
244 if format is None: 244 ↛ 246line 244 didn't jump to line 246 because the condition on line 244 was always true
245 format = source.content_type
246 could_not_guess_format = False
247 if format is None: 247 ↛ 258line 247 didn't jump to line 258 because the condition on line 247 was always true
248 _file = getattr(source, "file", None)
249 if ( 249 ↛ 255line 249 didn't jump to line 255 because the condition on line 249 was always true
250 _file is not None
251 and getattr(_file, "name", None)
252 and isinstance(_file.name, str)
253 ):
254 format = rdflib.util.guess_format(_file.name)
255 if format is None: 255 ↛ 256line 255 didn't jump to line 256 because the condition on line 255 was never true
256 format = "turtle"
257 could_not_guess_format = True
258 parser = plugin.get(format, Parser)()
259 try:
260 parser.parse(source, self, **args)
261 except SyntaxError as se:
262 if could_not_guess_format:
263 raise ParserError(
264 "Could not guess RDF format for %r from file"
265 " extension so tried Turtle but failed."
266 " You can explicitly specify format using"
267 " the format argument." % source
268 )
269 else:
270 raise se
271 finally:
272 if source.auto_close: 272 ↛ 275line 272 didn't jump to line 275 because the condition on line 272 was always true
273 source.close()
275 for subject in self.subjects(unique=True):
276 if subject not in self.all_entities: 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true
277 self.all_entities.add(subject)
279 if subject not in self.entity_index: 279 ↛ 280line 279 didn't jump to line 280 because the condition on line 279 was never true
280 self.entity_index[subject] = {
281 "to_be_deleted": False,
282 "is_restored": False,
283 "resp_agent": resp_agent,
284 "source": primary_source,
285 }
287 return self
290class OCDMDataset(OCDMGraphCommons, Dataset):
291 def __init__(self, counter_handler: CounterHandler | None = None):
292 Dataset.__init__(self)
293 self.preexisting_graph: Graph | Dataset = Dataset()
294 OCDMGraphCommons.__init__(self, counter_handler) # type: ignore[arg-type]
296 def __deepcopy__(self, memo):
297 new_graph = OCDMDataset(counter_handler=self.provenance.counter_handler)
299 # Copy graph data
300 for quad in self.quads((None, None, None, None)):
301 new_graph.add(quad) # type: ignore[arg-type]
303 # Copy entity index and metadata
304 for key, value in self.entity_index.items():
305 new_graph.entity_index[key] = value.copy()
306 new_graph.all_entities = self.all_entities.copy()
307 for key, value in self.merge_index.items(): 307 ↛ 308line 307 didn't jump to line 308 because the loop on line 307 never started
308 new_graph._OCDMGraphCommons__merge_index[key] = value.copy() # type: ignore[attr-defined]
310 return new_graph
312 def add( # type: ignore[override]
313 self,
314 triple_or_quad: tuple[Node, Node, Node] | tuple[Node, Node, Node, Graph | None],
315 resp_agent: object = None,
316 primary_source: object = None,
317 ) -> Dataset:
319 s, p, o, c = self._spoc(triple_or_quad, default=True)
321 _assertnode(s, p, o)
323 self.store.add(
324 (s, p, o),
325 context=c, # type: ignore[arg-type]
326 quoted=False,
327 )
329 # Add the subject to all_entities if it's not already present
330 if s not in self.all_entities:
331 self.all_entities.add(s)
333 if s not in self.entity_index:
334 self.entity_index[s] = {
335 "to_be_deleted": False,
336 "is_restored": False,
337 "resp_agent": resp_agent,
338 "source": primary_source,
339 "graph_iri": None,
340 }
342 # Store graph_iri in entity_index for later retrieval
343 # We already have the context from _spoc, use it directly for efficiency
344 if self.entity_index[s]["graph_iri"] is None:
345 self.entity_index[s]["graph_iri"] = _extract_graph_iri_from_context(c)
347 return self
349 def parse( # type: ignore[override]
350 self,
351 source: IO[bytes]
352 | TextIO
353 | InputSource
354 | str
355 | bytes
356 | pathlib.PurePath
357 | None = None,
358 publicID: str | None = None, # noqa: N803
359 format: str | None = None,
360 location: str | None = None,
361 file: BinaryIO | TextIO | None = None,
362 data: str | bytes | None = None,
363 resp_agent: URIRef | None = None,
364 primary_source: URIRef | None = None,
365 **args: object,
366 ) -> Graph:
367 source = create_input_source(
368 source=source,
369 publicID=publicID,
370 location=location,
371 file=file,
372 data=data,
373 format=format,
374 )
376 g_id = publicID or source.getPublicId() or ""
377 if not isinstance(g_id, Node): 377 ↛ 378line 377 didn't jump to line 378 because the condition on line 377 was never true
378 g_id = URIRef(g_id)
380 context = Graph(store=self.store, identifier=g_id)
381 context.remove((None, None, None)) # type: ignore[arg-type]
382 context.parse(source, publicID=publicID, format=format, **args) # type: ignore[arg-type]
383 # TODO: FIXME: This should not return context, but self.
385 unique_subjects = set()
386 for s, _, _, _ in self.quads((None, None, None, None)):
387 unique_subjects.add(s)
389 for subject in unique_subjects:
390 if subject not in self.all_entities: 390 ↛ 393line 390 didn't jump to line 393 because the condition on line 390 was always true
391 self.all_entities.add(subject)
393 if subject not in self.entity_index: 393 ↛ 403line 393 didn't jump to line 403 because the condition on line 393 was always true
394 self.entity_index[subject] = {
395 "to_be_deleted": False,
396 "is_restored": False,
397 "resp_agent": resp_agent,
398 "source": primary_source,
399 "graph_iri": None,
400 }
402 # Store graph_iri for this subject by finding its context
403 if ( 403 ↛ 389line 403 didn't jump to line 389 because the condition on line 403 was always true
404 "graph_iri" not in self.entity_index[subject]
405 or self.entity_index[subject]["graph_iri"] is None
406 ):
407 self.entity_index[subject]["graph_iri"] = _extract_graph_iri(
408 self, subject
409 )
411 return context
414def _assertnode(*terms):
415 for t in terms:
416 assert isinstance(t, Node), "Term %s must be an rdflib term" % (t,)
417 return True
420# Backward compatibility alias
421class OCDMConjunctiveGraph(OCDMDataset):
422 """
423 Deprecated: Use OCDMDataset instead.
425 This class is maintained for backward compatibility only.
426 OCDMConjunctiveGraph has been renamed to OCDMDataset to reflect
427 the migration from the deprecated ConjunctiveGraph to Dataset.
428 """
430 def __init__(self, counter_handler: CounterHandler | None = None):
431 warnings.warn(
432 "OCDMConjunctiveGraph is deprecated, use OCDMDataset instead",
433 DeprecationWarning,
434 stacklevel=2,
435 )
436 super().__init__(counter_handler)