Coverage for oc_ocdm / prov / prov_set.py: 95%
190 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-05-08 20:23 +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]] = {
35 "se": "snapshot of entity metadata"
36 }
38 def __init__(self, prov_subj_graph_set: GraphSet, base_iri: str, info_dir: str = "",
39 wanted_label: bool = True, custom_counter_handler: Optional[CounterHandler] = None,
40 supplier_prefix: str = "") -> None:
41 super(ProvSet, self).__init__()
42 self.prov_g: GraphSet = prov_subj_graph_set
43 self.res_to_entity: Dict[str, ProvEntity] = {}
44 self.base_iri: str = base_iri
45 self.wanted_label: bool = wanted_label
46 self.info_dir = info_dir
47 self.supplier_prefix = supplier_prefix
48 if custom_counter_handler:
49 self.counter_handler = custom_counter_handler
50 elif info_dir is not None and info_dir != "":
51 self.counter_handler = FilesystemCounterHandler(info_dir, supplier_prefix=supplier_prefix)
52 else:
53 self.counter_handler = InMemoryCounterHandler()
55 def get_entity(self, res: str) -> Optional[ProvEntity]:
56 if res in self.res_to_entity:
57 return self.res_to_entity[res]
59 def add_se(self, prov_subject: GraphEntity, res: Optional[str] = None) -> SnapshotEntity:
60 if res is not None and get_short_name(res) != "se":
61 raise ValueError(f"Given res: <{res}> is inappropriate for a SnapshotEntity entity.")
62 if res is not None and res in self.res_to_entity:
63 entity = self.res_to_entity[res]
64 assert isinstance(entity, SnapshotEntity)
65 return entity
66 g_prov: str = str(prov_subject) + "/prov/"
67 supplier_prefix = get_prefix(prov_subject.res)
68 cur_g, count, label = self._add_prov(g_prov, "se", prov_subject, res, supplier_prefix)
69 return SnapshotEntity(prov_subject, cur_g, self,
70 res, prov_subject.resp_agent, prov_subject.source, count, label)
72 def _create_snapshot(self, cur_subj: GraphEntity, cur_time: str) -> SnapshotEntity:
73 new_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj)
74 new_snapshot.is_snapshot_of(cur_subj)
75 new_snapshot.has_generation_time(cur_time)
76 if cur_subj.source is not None:
77 new_snapshot.has_primary_source(cur_subj.source)
78 if cur_subj.resp_agent is not None:
79 new_snapshot.has_resp_agent(cur_subj.resp_agent)
80 return new_snapshot
82 def _get_snapshots_from_merge_list(self, cur_subj: GraphEntity) -> List[SnapshotEntity]:
83 snapshots_list: List[SnapshotEntity] = []
84 for entity in cur_subj.merge_list:
85 last_entity_snapshot_res: Optional[str] = self._retrieve_last_snapshot(entity.res)
86 if last_entity_snapshot_res is not None:
87 snapshots_list.append(self.add_se(prov_subject=entity, res=last_entity_snapshot_res))
88 return snapshots_list
90 @staticmethod
91 def _get_merge_description(cur_subj: GraphEntity, snapshots_list: List[SnapshotEntity]) -> str:
92 merge_description: str = f"The entity '{cur_subj.res}' has been merged"
93 is_first: bool = True
94 for snapshot in snapshots_list:
95 if is_first:
96 merge_description += f" with '{snapshot.prov_subject.res}'"
97 is_first = False
98 else:
99 merge_description += f", '{snapshot.prov_subject.res}'"
100 merge_description += "."
101 return merge_description
103 def generate_provenance(self, c_time: Optional[float] = None) -> set:
104 modified_entities = set()
106 if c_time is None:
107 cur_time: str = datetime.now(tz=timezone.utc).replace(microsecond=0).isoformat(sep="T")
108 else:
109 cur_time: str = datetime.fromtimestamp(c_time, tz=timezone.utc).replace(microsecond=0).isoformat(sep="T")
111 # MERGED ENTITIES
112 for cur_subj in self.prov_g.res_to_entity.values():
113 if cur_subj is None or (not cur_subj.was_merged or cur_subj.to_be_deleted):
114 # Here we must skip every entity that was not merged or that must be deleted.
115 continue
117 # Previous snapshot
118 last_snapshot_res: Optional[str] = self._retrieve_last_snapshot(cur_subj.res)
119 if last_snapshot_res is None:
120 # CREATION SNAPSHOT
121 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
122 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been created.")
123 modified_entities.add(cur_subj.res)
124 else:
125 update_queries, _, _ = get_update_query(cur_subj, entity_type="graph")
126 was_modified: bool = len(update_queries) > 0
127 update_query: str = " ; ".join(update_queries) if update_queries else ""
128 snapshots_list: List[SnapshotEntity] = self._get_snapshots_from_merge_list(cur_subj)
129 if was_modified and len(snapshots_list) <= 0:
130 # MODIFICATION SNAPSHOT
131 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
132 last_snapshot.has_invalidation_time(cur_time)
134 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
135 cur_snapshot.derives_from(last_snapshot)
136 cur_snapshot.has_update_action(update_query)
137 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been modified.")
138 modified_entities.add(cur_subj.res)
139 elif len(snapshots_list) > 0:
140 # MERGE SNAPSHOT
141 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
142 last_snapshot.has_invalidation_time(cur_time)
143 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
144 cur_snapshot.derives_from(last_snapshot)
145 for snapshot in snapshots_list:
146 cur_snapshot.derives_from(snapshot)
147 if update_query:
148 cur_snapshot.has_update_action(update_query)
149 cur_snapshot.has_description(self._get_merge_description(cur_subj, snapshots_list))
150 modified_entities.add(cur_subj.res)
152 # EVERY OTHER ENTITY
153 for cur_subj in self.prov_g.res_to_entity.values():
154 if cur_subj is None or (cur_subj.was_merged and not cur_subj.to_be_deleted):
155 # Here we must skip every entity which was merged while not being marked as to be deleted,
156 # since we already processed those entities in the previous loop.
157 continue
159 last_snapshot_res: Optional[str] = self._retrieve_last_snapshot(cur_subj.res)
160 if last_snapshot_res is None:
161 if cur_subj.to_be_deleted:
162 # We can ignore this entity because it was deleted even before being created.
163 pass
164 else:
165 # CREATION SNAPSHOT
166 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
167 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been created.")
168 modified_entities.add(cur_subj.res)
169 else:
170 update_queries, _, _ = get_update_query(cur_subj, entity_type="graph")
171 was_modified: bool = len(update_queries) > 0
172 update_query: str = " ; ".join(update_queries) if update_queries else ""
173 if cur_subj.to_be_deleted:
174 # DELETION SNAPSHOT
175 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
176 last_snapshot.has_invalidation_time(cur_time)
178 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
179 cur_snapshot.derives_from(last_snapshot)
180 cur_snapshot.has_invalidation_time(cur_time)
181 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been deleted.")
182 cur_snapshot.has_update_action(update_query)
183 modified_entities.add(cur_subj.res)
184 elif cur_subj.is_restored:
185 # RESTORATION SNAPSHOT
186 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
187 # Don't set invalidation time on previous snapshot for restorations
189 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
190 cur_snapshot.derives_from(last_snapshot)
191 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been restored.")
192 if update_query:
193 cur_snapshot.has_update_action(update_query)
194 modified_entities.add(cur_subj.res)
195 elif was_modified:
196 # MODIFICATION SNAPSHOT
197 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
198 last_snapshot.has_invalidation_time(cur_time)
200 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
201 cur_snapshot.derives_from(last_snapshot)
202 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been modified.")
203 cur_snapshot.has_update_action(update_query)
204 modified_entities.add(cur_subj.res)
205 return modified_entities
207 def _add_prov(self, graph_url: str, short_name: str, prov_subject: GraphEntity,
208 res: Optional[str] = None, supplier_prefix: str = "") -> Tuple[TripleLite, Optional[str], Optional[str]]:
209 cur_g = TripleLite(identifier=graph_url)
211 count: Optional[str] = None
212 label: Optional[str] = None
214 if res is not None:
215 try:
216 res_count: int = int(get_count(res))
217 except ValueError:
218 res_count: int = -1
220 if isinstance(self.counter_handler, SqliteCounterHandler):
221 cur_count: int = self.counter_handler.read_counter(str(prov_subject))
222 else:
223 cur_count: int = self.counter_handler.read_counter(prov_subject.short_name, "se", int(get_count(prov_subject.res)), supplier_prefix=supplier_prefix)
225 if res_count > cur_count:
226 if isinstance(self.counter_handler, SqliteCounterHandler):
227 self.counter_handler.set_counter(int(get_count(prov_subject.res)), str(prov_subject))
228 else:
229 self.counter_handler.set_counter(res_count, prov_subject.short_name, "se", int(get_count(prov_subject.res)), supplier_prefix=supplier_prefix)
230 return cur_g, count, label
232 if isinstance(self.counter_handler, SqliteCounterHandler):
233 count = str(self.counter_handler.increment_counter(str(prov_subject)))
234 else:
235 count = str(self.counter_handler.increment_counter(prov_subject.short_name, "se", int(get_count(prov_subject.res)), supplier_prefix=supplier_prefix))
237 if self.wanted_label:
238 cur_short_name = prov_subject.short_name
239 cur_entity_count = get_count(prov_subject.res)
240 cur_entity_prefix = get_prefix(prov_subject.res)
242 related_to_label = "related to %s %s%s" % (GraphSet.labels[cur_short_name], cur_entity_prefix,
243 cur_entity_count)
244 related_to_short_label = "-> %s/%s%s" % (cur_short_name, cur_entity_prefix, cur_entity_count)
246 label = "%s %s %s [%s/%s %s]" % (self.labels[short_name], count, related_to_label, short_name, count,
247 related_to_short_label)
249 return cur_g, count, label
251 def _retrieve_last_snapshot(self, prov_subject: str) -> Optional[str]:
252 subj_short_name: str = get_short_name(prov_subject)
253 try:
254 subj_count: str = get_count(prov_subject)
255 if int(subj_count) <= 0:
256 raise ValueError('prov_subject is not a valid URIRef. Extracted count value should be a positive '
257 'non-zero integer number!')
258 except ValueError:
259 raise ValueError('prov_subject is not a valid URIRef. Unable to extract the count value!')
261 supplier_prefix = get_prefix(prov_subject)
263 if isinstance(self.counter_handler, SqliteCounterHandler):
264 last_snapshot_count: str = str(self.counter_handler.read_counter(str(prov_subject)))
265 else:
266 last_snapshot_count: str = str(self.counter_handler.read_counter(subj_short_name, "se", int(subj_count), supplier_prefix=supplier_prefix))
268 if int(last_snapshot_count) <= 0:
269 return None
270 else:
271 return str(prov_subject) + '/prov/se/' + last_snapshot_count
273 def get_se(self) -> Tuple[SnapshotEntity, ...]:
274 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, SnapshotEntity))