Coverage for oc_ocdm / prov / prov_set.py: 95%
190 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-06 20:05 +0000
1#!/usr/bin/python
3# SPDX-FileCopyrightText: 2022-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
4#
5# SPDX-License-Identifier: ISC
7# -*- coding: utf-8 -*-
8from __future__ import annotations
10from datetime import datetime, timezone
11from typing import TYPE_CHECKING
13from oc_ocdm.abstract_set import AbstractSet
14from oc_ocdm.prov.entities.snapshot_entity import SnapshotEntity
15from oc_ocdm.support.query_utils import get_update_query
17if TYPE_CHECKING:
18 from typing import ClassVar, Dict, List, Optional, Tuple
20 from oc_ocdm.graph.graph_entity import GraphEntity
22from triplelite import TripleLite
24from oc_ocdm.counter_handler.counter_handler import CounterHandler
25from oc_ocdm.counter_handler.filesystem_counter_handler import FilesystemCounterHandler
26from oc_ocdm.counter_handler.in_memory_counter_handler import InMemoryCounterHandler
27from oc_ocdm.counter_handler.sqlite_counter_handler import SqliteCounterHandler
28from oc_ocdm.graph.graph_set import GraphSet
29from oc_ocdm.prov.prov_entity import ProvEntity
30from oc_ocdm.support.support import get_count, get_prefix, get_short_name
33class ProvSet(AbstractSet[ProvEntity]):
34 labels: ClassVar[Dict[str, str]] = {"se": "snapshot of entity metadata"}
36 def __init__(
37 self,
38 prov_subj_graph_set: GraphSet,
39 base_iri: str,
40 info_dir: str | None = "",
41 wanted_label: bool = True,
42 custom_counter_handler: Optional[CounterHandler] = None,
43 supplier_prefix: str = "",
44 ) -> None:
45 super(ProvSet, self).__init__()
46 self.prov_g: GraphSet = prov_subj_graph_set
47 self.res_to_entity: Dict[str, ProvEntity] = {}
48 self.base_iri: str = base_iri
49 self.wanted_label: bool = wanted_label
50 self.info_dir = info_dir
51 self.supplier_prefix = supplier_prefix
52 if custom_counter_handler:
53 self.counter_handler = custom_counter_handler
54 elif info_dir is not None and info_dir != "":
55 self.counter_handler = FilesystemCounterHandler(info_dir, supplier_prefix=supplier_prefix)
56 else:
57 self.counter_handler = InMemoryCounterHandler()
59 def get_entity(self, res: str) -> Optional[ProvEntity]:
60 if res in self.res_to_entity:
61 return self.res_to_entity[res]
63 def add_se(self, prov_subject: GraphEntity, res: Optional[str] = None) -> SnapshotEntity:
64 if res is not None and get_short_name(res) != "se":
65 raise ValueError(f"Given res: <{res}> is inappropriate for a SnapshotEntity entity.")
66 if res is not None and res in self.res_to_entity:
67 entity = self.res_to_entity[res]
68 assert isinstance(entity, SnapshotEntity)
69 return entity
70 g_prov: str = str(prov_subject) + "/prov/"
71 supplier_prefix = get_prefix(prov_subject.res)
72 cur_g, count, label = self._add_prov(g_prov, "se", prov_subject, res, supplier_prefix)
73 return SnapshotEntity(
74 prov_subject, cur_g, self, res, prov_subject.resp_agent, prov_subject.source, count, label
75 )
77 def _create_snapshot(self, cur_subj: GraphEntity, cur_time: str) -> SnapshotEntity:
78 new_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj)
79 new_snapshot.is_snapshot_of(cur_subj)
80 new_snapshot.has_generation_time(cur_time)
81 if cur_subj.source is not None:
82 new_snapshot.has_primary_source(cur_subj.source)
83 if cur_subj.resp_agent is not None:
84 new_snapshot.has_resp_agent(cur_subj.resp_agent)
85 return new_snapshot
87 def _get_snapshots_from_merge_list(self, cur_subj: GraphEntity) -> List[SnapshotEntity]:
88 snapshots_list: List[SnapshotEntity] = []
89 for entity in cur_subj.merge_list:
90 last_entity_snapshot_res: Optional[str] = self._retrieve_last_snapshot(entity.res)
91 if last_entity_snapshot_res is not None:
92 snapshots_list.append(self.add_se(prov_subject=entity, res=last_entity_snapshot_res))
93 return snapshots_list
95 @staticmethod
96 def _get_merge_description(cur_subj: GraphEntity, snapshots_list: List[SnapshotEntity]) -> str:
97 merge_description: str = f"The entity '{cur_subj.res}' has been merged"
98 is_first: bool = True
99 for snapshot in snapshots_list:
100 if is_first:
101 merge_description += f" with '{snapshot.prov_subject.res}'"
102 is_first = False
103 else:
104 merge_description += f", '{snapshot.prov_subject.res}'"
105 merge_description += "."
106 return merge_description
108 def generate_provenance(self, c_time: Optional[float] = None) -> set[str]:
109 modified_entities: set[str] = set()
111 if c_time is None:
112 cur_time: str = datetime.now(tz=timezone.utc).replace(microsecond=0).isoformat(sep="T")
113 else:
114 cur_time: str = datetime.fromtimestamp(c_time, tz=timezone.utc).replace(microsecond=0).isoformat(sep="T")
116 # MERGED ENTITIES
117 for cur_subj in self.prov_g.res_to_entity.values():
118 if not cur_subj.was_merged or cur_subj.to_be_deleted:
119 # Here we must skip every entity that was not merged or that must be deleted.
120 continue
122 # Previous snapshot
123 last_snapshot_res: Optional[str] = self._retrieve_last_snapshot(cur_subj.res)
124 if last_snapshot_res is None:
125 # CREATION SNAPSHOT
126 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
127 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been created.")
128 modified_entities.add(cur_subj.res)
129 else:
130 update_queries, _, _ = get_update_query(cur_subj, entity_type="graph")
131 was_modified: bool = len(update_queries) > 0
132 update_query: str = " ; ".join(update_queries) if update_queries else ""
133 snapshots_list: List[SnapshotEntity] = self._get_snapshots_from_merge_list(cur_subj)
134 if was_modified and len(snapshots_list) <= 0:
135 # MODIFICATION SNAPSHOT
136 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
137 last_snapshot.has_invalidation_time(cur_time)
139 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
140 cur_snapshot.derives_from(last_snapshot)
141 cur_snapshot.has_update_action(update_query)
142 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been modified.")
143 modified_entities.add(cur_subj.res)
144 elif len(snapshots_list) > 0:
145 # MERGE SNAPSHOT
146 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
147 last_snapshot.has_invalidation_time(cur_time)
148 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
149 cur_snapshot.derives_from(last_snapshot)
150 for snapshot in snapshots_list:
151 cur_snapshot.derives_from(snapshot)
152 if update_query:
153 cur_snapshot.has_update_action(update_query)
154 cur_snapshot.has_description(self._get_merge_description(cur_subj, snapshots_list))
155 modified_entities.add(cur_subj.res)
157 # EVERY OTHER ENTITY
158 for cur_subj in self.prov_g.res_to_entity.values():
159 if cur_subj.was_merged and not cur_subj.to_be_deleted:
160 # Here we must skip every entity which was merged while not being marked as to be deleted,
161 # since we already processed those entities in the previous loop.
162 continue
164 last_snapshot_res: Optional[str] = self._retrieve_last_snapshot(cur_subj.res)
165 if last_snapshot_res is None:
166 if cur_subj.to_be_deleted:
167 # We can ignore this entity because it was deleted even before being created.
168 pass
169 else:
170 # CREATION SNAPSHOT
171 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
172 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been created.")
173 modified_entities.add(cur_subj.res)
174 else:
175 update_queries, _, _ = get_update_query(cur_subj, entity_type="graph")
176 was_modified: bool = len(update_queries) > 0
177 update_query: str = " ; ".join(update_queries) if update_queries else ""
178 if cur_subj.to_be_deleted:
179 # DELETION SNAPSHOT
180 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
181 last_snapshot.has_invalidation_time(cur_time)
183 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
184 cur_snapshot.derives_from(last_snapshot)
185 cur_snapshot.has_invalidation_time(cur_time)
186 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been deleted.")
187 cur_snapshot.has_update_action(update_query)
188 modified_entities.add(cur_subj.res)
189 elif cur_subj.is_restored:
190 # RESTORATION SNAPSHOT
191 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
192 # Don't set invalidation time on previous snapshot for restorations
194 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
195 cur_snapshot.derives_from(last_snapshot)
196 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been restored.")
197 if update_query:
198 cur_snapshot.has_update_action(update_query)
199 modified_entities.add(cur_subj.res)
200 elif was_modified:
201 # MODIFICATION SNAPSHOT
202 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
203 last_snapshot.has_invalidation_time(cur_time)
205 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
206 cur_snapshot.derives_from(last_snapshot)
207 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been modified.")
208 cur_snapshot.has_update_action(update_query)
209 modified_entities.add(cur_subj.res)
210 return modified_entities
212 def _add_prov(
213 self,
214 graph_url: str,
215 short_name: str,
216 prov_subject: GraphEntity,
217 res: Optional[str] = None,
218 supplier_prefix: str = "",
219 ) -> Tuple[TripleLite, Optional[str], Optional[str]]:
220 cur_g = TripleLite(identifier=graph_url)
222 count: Optional[str] = None
223 label: Optional[str] = None
225 if res is not None:
226 try:
227 res_count: int = int(get_count(res))
228 except ValueError:
229 res_count: int = -1
231 if isinstance(self.counter_handler, SqliteCounterHandler):
232 cur_count: int = self.counter_handler.read_counter(str(prov_subject))
233 else:
234 cur_count: int = self.counter_handler.read_counter(
235 prov_subject.short_name, "se", int(get_count(prov_subject.res)), supplier_prefix=supplier_prefix
236 )
238 if res_count > cur_count:
239 if isinstance(self.counter_handler, SqliteCounterHandler):
240 self.counter_handler.set_counter(int(get_count(prov_subject.res)), str(prov_subject))
241 else:
242 self.counter_handler.set_counter(
243 res_count,
244 prov_subject.short_name,
245 "se",
246 int(get_count(prov_subject.res)),
247 supplier_prefix=supplier_prefix,
248 )
249 return cur_g, count, label
251 if isinstance(self.counter_handler, SqliteCounterHandler):
252 count = str(self.counter_handler.increment_counter(str(prov_subject)))
253 else:
254 count = str(
255 self.counter_handler.increment_counter(
256 prov_subject.short_name, "se", int(get_count(prov_subject.res)), supplier_prefix=supplier_prefix
257 )
258 )
260 if self.wanted_label:
261 cur_short_name = prov_subject.short_name
262 cur_entity_count = get_count(prov_subject.res)
263 cur_entity_prefix = get_prefix(prov_subject.res)
265 related_to_label = "related to %s %s%s" % (
266 GraphSet.labels[cur_short_name],
267 cur_entity_prefix,
268 cur_entity_count,
269 )
270 related_to_short_label = "-> %s/%s%s" % (cur_short_name, cur_entity_prefix, cur_entity_count)
272 label = "%s %s %s [%s/%s %s]" % (
273 self.labels[short_name],
274 count,
275 related_to_label,
276 short_name,
277 count,
278 related_to_short_label,
279 )
281 return cur_g, count, label
283 def _retrieve_last_snapshot(self, prov_subject: str) -> Optional[str]:
284 subj_short_name: str = get_short_name(prov_subject)
285 try:
286 subj_count: str = get_count(prov_subject)
287 if int(subj_count) <= 0:
288 raise ValueError(
289 "prov_subject is not a valid URIRef. Extracted count value should be a positive "
290 "non-zero integer number!"
291 )
292 except ValueError:
293 raise ValueError("prov_subject is not a valid URIRef. Unable to extract the count value!")
295 supplier_prefix = get_prefix(prov_subject)
297 if isinstance(self.counter_handler, SqliteCounterHandler):
298 last_snapshot_count: str = str(self.counter_handler.read_counter(str(prov_subject)))
299 else:
300 last_snapshot_count: str = str(
301 self.counter_handler.read_counter(
302 subj_short_name, "se", int(subj_count), supplier_prefix=supplier_prefix
303 )
304 )
306 if int(last_snapshot_count) <= 0:
307 return None
308 else:
309 return str(prov_subject) + "/prov/se/" + last_snapshot_count
311 def get_se(self) -> Tuple[SnapshotEntity, ...]:
312 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, SnapshotEntity))