Coverage for heritrace/editor.py: 98%
175 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-26 08:34 +0000
1# SPDX-FileCopyrightText: 2024-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5from dataclasses import dataclass
6from datetime import datetime, timezone
7from typing import TYPE_CHECKING
9from rdflib import Literal, URIRef
10from rdflib_ocdm.counter_handler.counter_handler import CounterHandler
11from rdflib_ocdm.ocdm_graph import OCDMDataset, OCDMGraph
12from rdflib_ocdm.reader import Reader
13from rdflib_ocdm.storer import Storer
14from SPARQLWrapper import JSON
16from heritrace.counter_handler import TransactionalCounterHandler
17from heritrace.sparql import SPARQLWrapperWithRetry, get_sparql_bindings
19if TYPE_CHECKING:
20 from heritrace.save_plugin import SavePlugin
23@dataclass(frozen=True, slots=True)
24class EndpointConfig:
25 dataset: str
26 provenance: str
27 is_quadstore: bool = True
30class EditorError(Exception):
31 pass
34class Editor:
35 def __init__( # noqa: PLR0913
36 self,
37 endpoints: EndpointConfig,
38 counter_handler: CounterHandler,
39 resp_agent: URIRef,
40 source: URIRef | None = None,
41 c_time: datetime | None = None,
42 save_plugin: "SavePlugin | None" = None,
43 ) -> None:
44 self.dataset_endpoint = endpoints.dataset
45 self.provenance_endpoint = endpoints.provenance
46 self.counter_handler = counter_handler
47 self.resp_agent = resp_agent
48 self.source = source
49 self.c_time = self.to_posix_timestamp(c_time)
50 self.save_plugin = save_plugin
51 self.dataset_is_quadstore = endpoints.is_quadstore
52 self.transactional_counter_handler: TransactionalCounterHandler | None = (
53 counter_handler
54 if isinstance(counter_handler, TransactionalCounterHandler)
55 else None
56 )
57 self._counter_transaction_started = False
58 self.g_set = (
59 OCDMDataset(self.counter_handler)
60 if self.dataset_is_quadstore
61 else OCDMGraph(self.counter_handler)
62 )
63 self.begin_counter_transaction()
65 def create(
66 self,
67 subject: URIRef,
68 predicate: URIRef,
69 value: Literal | URIRef,
70 graph: URIRef | None = None,
71 ) -> None:
72 if self.dataset_is_quadstore and graph:
73 self.g_set.add( # type: ignore[arg-type]
74 (subject, predicate, value, graph), # type: ignore[arg-type]
75 resp_agent=self.resp_agent,
76 primary_source=self.source,
77 )
78 else:
79 self.g_set.add( # type: ignore[arg-type]
80 (subject, predicate, value),
81 resp_agent=self.resp_agent,
82 primary_source=self.source,
83 )
85 def update(
86 self,
87 subject: URIRef,
88 predicate: URIRef,
89 old_value: Literal | URIRef,
90 new_value: Literal | URIRef,
91 graph: URIRef | None = None,
92 ) -> None:
93 if self.dataset_is_quadstore and graph:
94 if (subject, predicate, old_value, graph) not in self.g_set: # type: ignore[operator]
95 msg = (
96 f"Triple ({subject}, {predicate},"
97 f" {old_value}, {graph}) does not exist"
98 )
99 raise EditorError(msg)
100 self.g_set.remove((subject, predicate, old_value, graph)) # type: ignore[arg-type]
101 self.g_set.add( # type: ignore[arg-type]
102 (subject, predicate, new_value, graph), # type: ignore[arg-type]
103 resp_agent=self.resp_agent,
104 primary_source=self.source,
105 )
106 else:
107 if (subject, predicate, old_value) not in self.g_set: # type: ignore[operator]
108 msg = f"Triple ({subject}, {predicate}, {old_value}) does not exist"
109 raise EditorError(msg)
110 self.g_set.remove((subject, predicate, old_value)) # type: ignore[arg-type]
111 self.g_set.add( # type: ignore[arg-type]
112 (subject, predicate, new_value),
113 resp_agent=self.resp_agent,
114 primary_source=self.source,
115 )
117 def _delete_full_entity(self, subject: URIRef) -> None:
118 if self.dataset_is_quadstore:
119 quads = list(self.g_set.quads((subject, None, None, None))) # type: ignore[arg-type]
120 if not quads:
121 msg = f"Entity {subject} does not exist"
122 raise EditorError(msg)
123 for quad in quads:
124 self.g_set.remove(quad) # type: ignore[arg-type]
126 object_quads = list(self.g_set.quads((None, None, subject, None))) # type: ignore[arg-type]
127 for quad in object_quads:
128 self.g_set.remove(quad) # type: ignore[arg-type]
129 else:
130 triples = list(self.g_set.triples((subject, None, None))) # type: ignore[arg-type]
131 if not triples:
132 msg = f"Entity {subject} does not exist"
133 raise EditorError(msg)
134 for triple in triples:
135 self.g_set.remove(triple) # type: ignore[arg-type]
137 object_triples = list(self.g_set.triples((None, None, subject))) # type: ignore[arg-type]
138 for triple in object_triples:
139 self.g_set.remove(triple) # type: ignore[arg-type]
140 self.g_set.mark_as_deleted(subject) # type: ignore[arg-type]
142 def _delete_specific_triple(
143 self,
144 subject: URIRef,
145 predicate: URIRef,
146 value: Literal | URIRef,
147 graph: URIRef | None,
148 ) -> None:
149 if self.dataset_is_quadstore and graph:
150 if (subject, predicate, value, graph) not in self.g_set: # type: ignore[operator]
151 msg = (
152 f"Triple ({subject}, {predicate}, {value}, {graph}) does not exist"
153 )
154 raise EditorError(msg)
155 self.g_set.remove((subject, predicate, value, graph)) # type: ignore[arg-type]
156 else:
157 if (subject, predicate, value) not in self.g_set: # type: ignore[operator]
158 msg = f"Triple ({subject}, {predicate}, {value}) does not exist"
159 raise EditorError(msg)
160 self.g_set.remove((subject, predicate, value)) # type: ignore[arg-type]
162 def _delete_all_for_predicate(
163 self,
164 subject: URIRef,
165 predicate: URIRef,
166 graph: URIRef | None,
167 ) -> None:
168 if self.dataset_is_quadstore and graph:
169 quads = list(self.g_set.quads((subject, predicate, None, graph))) # type: ignore[arg-type]
170 if not quads:
171 msg = (
172 f"No triples found with subject"
173 f" {subject} and predicate"
174 f" {predicate} in graph {graph}"
175 )
176 raise EditorError(msg)
177 for quad in quads:
178 self.g_set.remove(quad) # type: ignore[arg-type]
179 else:
180 triples = list(self.g_set.triples((subject, predicate, None))) # type: ignore[arg-type]
181 if not triples:
182 msg = (
183 f"No triples found with subject {subject} and predicate {predicate}"
184 )
185 raise EditorError(msg)
186 for triple in triples:
187 self.g_set.remove(triple) # type: ignore[arg-type]
189 def delete(
190 self,
191 subject: URIRef,
192 predicate: URIRef | None = None,
193 value: Literal | URIRef | None = None,
194 graph: URIRef | None = None,
195 ) -> None:
196 if predicate is None:
197 self._delete_full_entity(subject)
198 elif value:
199 self._delete_specific_triple(subject, predicate, value, graph)
200 else:
201 self._delete_all_for_predicate(subject, predicate, graph)
203 from heritrace.utils.sparql_utils import get_triples_from_graph # noqa: PLC0415
205 if len(list(get_triples_from_graph(self.g_set, (subject, None, None)))) == 0:
206 self.g_set.mark_as_deleted(subject) # type: ignore[arg-type]
208 def import_entity(self, subject: URIRef) -> None:
209 Reader.import_entities_from_triplestore(
210 self.g_set,
211 self.dataset_endpoint,
212 [subject], # type: ignore[arg-type]
213 )
215 def merge(
216 self,
217 keep_entity_uri: URIRef,
218 delete_entity_uri: URIRef,
219 primary_source: URIRef | None = None,
220 ) -> None:
221 if keep_entity_uri == delete_entity_uri:
222 msg = "Cannot merge an entity with itself."
223 raise ValueError(msg)
225 merge_sparql = SPARQLWrapperWithRetry(self.dataset_endpoint)
226 entities_to_import: set[URIRef] = {keep_entity_uri, delete_entity_uri}
228 query_incoming = (
229 "SELECT DISTINCT ?s WHERE {"
230 f" ?s ?p <{delete_entity_uri}> ."
231 f" FILTER (?s != <{keep_entity_uri}>) }}"
232 )
233 merge_sparql.setQuery(query_incoming)
234 merge_sparql.setReturnFormat(JSON)
235 for binding in get_sparql_bindings(merge_sparql.query().convert()):
236 s_uri = URIRef(binding["s"]["value"])
237 entities_to_import.add(s_uri)
239 Reader.import_entities_from_triplestore(
240 self.g_set,
241 self.dataset_endpoint,
242 list(entities_to_import), # type: ignore[arg-type]
243 )
244 self.begin_counter_transaction()
245 self.g_set.preexisting_finished(self.resp_agent, self.source, self.c_time) # type: ignore[arg-type]
246 self.set_primary_source(primary_source)
247 self.g_set.merge(keep_entity_uri, delete_entity_uri) # type: ignore[arg-type]
249 self.save()
251 def preexisting_finished(self) -> None:
252 self.begin_counter_transaction()
253 self.g_set.preexisting_finished(self.resp_agent, self.source, self.c_time) # type: ignore[arg-type]
255 def save(self) -> None:
256 self.begin_counter_transaction()
257 try:
258 self.g_set.generate_provenance() # type: ignore[arg-type]
259 dataset_storer = Storer(self.g_set) # type: ignore[arg-type]
260 prov_storer = Storer(self.g_set.provenance) # type: ignore[attr-defined]
261 self._upload_or_raise(
262 dataset_storer,
263 self.dataset_endpoint,
264 "Failed to update the dataset triplestore",
265 )
266 self._upload_or_raise(
267 prov_storer,
268 self.provenance_endpoint,
269 "Failed to update the provenance triplestore",
270 )
271 if self.save_plugin is not None:
272 self.save_plugin.persist(self.g_set)
273 self.g_set.commit_changes() # type: ignore[arg-type]
274 self._commit_counter_transaction()
275 finally:
276 if self._counter_transaction_started:
277 self._rollback_counter_transaction()
279 @staticmethod
280 def _upload_or_raise(storer: Storer, endpoint: str, error_message: str) -> None:
281 if not storer.upload_all(endpoint): # type: ignore[arg-type]
282 raise EditorError(error_message)
284 def begin_counter_transaction(self) -> None:
285 if (
286 self._counter_transaction_started
287 or self.transactional_counter_handler is None
288 ):
289 return
290 self.transactional_counter_handler.begin_counter_transaction()
291 self._counter_transaction_started = True
293 def _commit_counter_transaction(self) -> None:
294 if (
295 not self._counter_transaction_started
296 or self.transactional_counter_handler is None
297 ):
298 return
299 self.transactional_counter_handler.commit_counter_transaction()
300 self._counter_transaction_started = False
302 def _rollback_counter_transaction(self) -> None:
303 if (
304 not self._counter_transaction_started
305 or self.transactional_counter_handler is None
306 ):
307 return
308 self.transactional_counter_handler.rollback_counter_transaction()
309 self._counter_transaction_started = False
311 def to_posix_timestamp(self, value: str | datetime | None) -> float | None:
312 if value is None:
313 return None
314 if isinstance(value, datetime):
315 return value.timestamp()
316 if isinstance(value, str):
317 dt = datetime.fromisoformat(value)
318 if dt.tzinfo is None:
319 dt = dt.replace(tzinfo=timezone.utc)
320 return dt.timestamp()
321 return None
323 def set_primary_source(self, source: URIRef | None) -> None:
324 self.source = source
325 for metadata in self.g_set.entity_index.values(): # type: ignore[union-attr]
326 metadata["source"] = source