Coverage for oc_ocdm/prov/prov_set.py: 94%
196 statements
« prev ^ index » next coverage.py v6.5.0, created at 2025-12-05 23:58 +0000
1#!/usr/bin/python
2# -*- coding: utf-8 -*-
3# Copyright (c) 2016, Silvio Peroni <essepuntato@gmail.com>
4#
5# Permission to use, copy, modify, and/or distribute this software for any purpose
6# with or without fee is hereby granted, provided that the above copyright notice
7# and this permission notice appear in all copies.
8#
9# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
10# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND
11# FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT,
12# OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE,
13# DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
14# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS
15# SOFTWARE.
16from __future__ import annotations
18import os
19from datetime import datetime, timezone
20from typing import TYPE_CHECKING
22from oc_ocdm.abstract_set import AbstractSet
23from oc_ocdm.prov.entities.snapshot_entity import SnapshotEntity
24from oc_ocdm.support.query_utils import get_update_query
26if TYPE_CHECKING:
27 from typing import Optional, Tuple, List, Dict, ClassVar
28 from oc_ocdm.graph.graph_entity import GraphEntity
30from rdflib import Graph, URIRef
32from oc_ocdm.counter_handler.counter_handler import CounterHandler
33from oc_ocdm.counter_handler.filesystem_counter_handler import \
34 FilesystemCounterHandler
35from oc_ocdm.counter_handler.in_memory_counter_handler import \
36 InMemoryCounterHandler
37from oc_ocdm.counter_handler.sqlite_counter_handler import SqliteCounterHandler
38from oc_ocdm.graph.graph_set import GraphSet
39from oc_ocdm.prov.prov_entity import ProvEntity
40from oc_ocdm.support.support import (get_count, get_prefix, get_short_name)
43class ProvSet(AbstractSet):
44 labels: ClassVar[Dict[str, str]] = {
45 "se": "snapshot of entity metadata"
46 }
48 def __init__(self, prov_subj_graph_set: GraphSet, base_iri: str, info_dir: str = "",
49 wanted_label: bool = True, custom_counter_handler: CounterHandler = None,
50 supplier_prefix: str = "") -> None:
51 super(ProvSet, self).__init__()
52 self.prov_g: GraphSet = prov_subj_graph_set
53 # The following variable maps a URIRef with the related provenance entity
54 self.res_to_entity: Dict[URIRef, ProvEntity] = {}
55 self.base_iri: str = base_iri
56 self.wanted_label: bool = wanted_label
57 self.info_dir = info_dir
58 self.supplier_prefix = supplier_prefix
59 if custom_counter_handler:
60 self.counter_handler = custom_counter_handler
61 elif info_dir is not None and info_dir != "":
62 self.counter_handler = FilesystemCounterHandler(info_dir, supplier_prefix=supplier_prefix)
63 else:
64 self.counter_handler = InMemoryCounterHandler()
66 def get_entity(self, res: URIRef) -> Optional[ProvEntity]:
67 if res in self.res_to_entity:
68 return self.res_to_entity[res]
70 def add_se(self, prov_subject: GraphEntity, res: URIRef = None) -> SnapshotEntity:
71 if res is not None and get_short_name(res) != "se":
72 raise ValueError(f"Given res: <{res}> is inappropriate for a SnapshotEntity entity.")
73 if res is not None and res in self.res_to_entity:
74 return self.res_to_entity[res]
75 g_prov: str = str(prov_subject) + "/prov/"
76 supplier_prefix = get_prefix(str(prov_subject.res))
77 cur_g, count, label = self._add_prov(g_prov, "se", prov_subject, res, supplier_prefix)
78 return SnapshotEntity(prov_subject, cur_g, self, res, prov_subject.resp_agent,
79 prov_subject.source, ProvEntity.iri_entity, count, label, "se")
81 def _create_snapshot(self, cur_subj: GraphEntity, cur_time: str) -> SnapshotEntity:
82 new_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj)
83 new_snapshot.is_snapshot_of(cur_subj)
84 new_snapshot.has_generation_time(cur_time)
85 if cur_subj.source is not None:
86 new_snapshot.has_primary_source(URIRef(cur_subj.source))
87 if cur_subj.resp_agent is not None:
88 new_snapshot.has_resp_agent(URIRef(cur_subj.resp_agent))
89 return new_snapshot
91 def _get_snapshots_from_merge_list(self, cur_subj: GraphEntity) -> List[SnapshotEntity]:
92 snapshots_list: List[SnapshotEntity] = []
93 for entity in cur_subj.merge_list:
94 last_entity_snapshot_res: Optional[URIRef] = self._retrieve_last_snapshot(entity.res)
95 if last_entity_snapshot_res is not None:
96 snapshots_list.append(self.add_se(prov_subject=entity, res=last_entity_snapshot_res))
97 return snapshots_list
99 @staticmethod
100 def _get_merge_description(cur_subj: GraphEntity, snapshots_list: List[SnapshotEntity]) -> str:
101 merge_description: str = f"The entity '{cur_subj.res}' has been merged"
102 is_first: bool = True
103 for snapshot in snapshots_list:
104 if is_first:
105 merge_description += f" with '{snapshot.prov_subject.res}'"
106 is_first = False
107 else:
108 merge_description += f", '{snapshot.prov_subject.res}'"
109 merge_description += "."
110 return merge_description
112 def generate_provenance(self, c_time: float = None) -> set:
113 modified_entities = set()
115 if c_time is None:
116 cur_time: str = datetime.now(tz=timezone.utc).replace(microsecond=0).isoformat(sep="T")
117 else:
118 cur_time: str = datetime.fromtimestamp(c_time, tz=timezone.utc).replace(microsecond=0).isoformat(sep="T")
120 # MERGED ENTITIES
121 for cur_subj in self.prov_g.res_to_entity.values():
122 if cur_subj is None or (not cur_subj.was_merged or cur_subj.to_be_deleted):
123 # Here we must skip every entity that was not merged or that must be deleted.
124 continue
126 # Previous snapshot
127 last_snapshot_res: Optional[URIRef] = self._retrieve_last_snapshot(cur_subj.res)
128 if last_snapshot_res is None:
129 # CREATION SNAPSHOT
130 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
131 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been created.")
132 modified_entities.add(cur_subj.res)
133 else:
134 update_queries, _, _ = get_update_query(cur_subj, entity_type="graph")
135 was_modified: bool = len(update_queries) > 0
136 update_query: str = " ; ".join(update_queries) if update_queries else ""
137 snapshots_list: List[SnapshotEntity] = self._get_snapshots_from_merge_list(cur_subj)
138 if was_modified and len(snapshots_list) <= 0:
139 # MODIFICATION SNAPSHOT
140 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
141 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 cur_snapshot.has_update_action(update_query)
146 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been modified.")
147 modified_entities.add(cur_subj.res)
148 elif len(snapshots_list) > 0:
149 # MERGE SNAPSHOT
150 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
151 last_snapshot.has_invalidation_time(cur_time)
152 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
153 cur_snapshot.derives_from(last_snapshot)
154 for snapshot in snapshots_list:
155 cur_snapshot.derives_from(snapshot)
156 if update_query:
157 cur_snapshot.has_update_action(update_query)
158 cur_snapshot.has_description(self._get_merge_description(cur_subj, snapshots_list))
159 modified_entities.add(cur_subj.res)
161 # EVERY OTHER ENTITY
162 for cur_subj in self.prov_g.res_to_entity.values():
163 if cur_subj is None or (cur_subj.was_merged and not cur_subj.to_be_deleted):
164 # Here we must skip every entity which was merged while not being marked as to be deleted,
165 # since we already processed those entities in the previous loop.
166 continue
168 last_snapshot_res: Optional[URIRef] = self._retrieve_last_snapshot(cur_subj.res)
169 if last_snapshot_res is None:
170 if cur_subj.to_be_deleted:
171 # We can ignore this entity because it was deleted even before being created.
172 pass
173 else:
174 # CREATION SNAPSHOT
175 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
176 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been created.")
177 modified_entities.add(cur_subj.res)
178 else:
179 update_queries, _, _ = get_update_query(cur_subj, entity_type="graph")
180 was_modified: bool = len(update_queries) > 0
181 update_query: str = " ; ".join(update_queries) if update_queries else ""
182 if cur_subj.to_be_deleted:
183 # DELETION SNAPSHOT
184 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
185 last_snapshot.has_invalidation_time(cur_time)
187 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
188 cur_snapshot.derives_from(last_snapshot)
189 cur_snapshot.has_invalidation_time(cur_time)
190 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been deleted.")
191 cur_snapshot.has_update_action(update_query)
192 modified_entities.add(cur_subj.res)
193 elif cur_subj.is_restored:
194 # RESTORATION SNAPSHOT
195 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
196 # Don't set invalidation time on previous snapshot for restorations
198 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
199 cur_snapshot.derives_from(last_snapshot)
200 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been restored.")
201 if update_query:
202 cur_snapshot.has_update_action(update_query)
203 modified_entities.add(cur_subj.res)
204 elif was_modified:
205 # MODIFICATION SNAPSHOT
206 last_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj, res=last_snapshot_res)
207 last_snapshot.has_invalidation_time(cur_time)
209 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
210 cur_snapshot.derives_from(last_snapshot)
211 cur_snapshot.has_description(f"The entity '{cur_subj.res}' has been modified.")
212 cur_snapshot.has_update_action(update_query)
213 modified_entities.add(cur_subj.res)
214 return modified_entities
216 def _add_prov(self, graph_url: str, short_name: str, prov_subject: GraphEntity,
217 res: URIRef = None, supplier_prefix: str = "") -> Tuple[Graph, Optional[str], Optional[str]]:
218 cur_g: Graph = Graph(identifier=graph_url)
219 self._set_ns(cur_g)
221 count: Optional[str] = None
222 label: Optional[str] = None
224 if res is not None:
225 try:
226 res_count: int = int(get_count(res))
227 except ValueError:
228 res_count: int = -1
230 if isinstance(self.counter_handler, SqliteCounterHandler):
231 cur_count: int = self.counter_handler.read_counter(prov_subject)
232 else:
233 cur_count: int = self.counter_handler.read_counter(prov_subject.short_name, "se", int(get_count(prov_subject.res)), supplier_prefix=supplier_prefix)
235 if res_count > cur_count:
236 if isinstance(self.counter_handler, SqliteCounterHandler):
237 self.counter_handler.set_counter(int(get_count(prov_subject.res)), prov_subject)
238 else:
239 self.counter_handler.set_counter(res_count, prov_subject.short_name, "se", int(get_count(prov_subject.res)), supplier_prefix=supplier_prefix)
240 return cur_g, count, label
242 if isinstance(self.counter_handler, SqliteCounterHandler):
243 count = str(self.counter_handler.increment_counter(prov_subject))
244 else:
245 count = str(self.counter_handler.increment_counter(prov_subject.short_name, "se", int(get_count(prov_subject.res)), supplier_prefix=supplier_prefix))
247 if self.wanted_label:
248 cur_short_name = prov_subject.short_name
249 cur_entity_count = get_count(prov_subject.res)
250 cur_entity_prefix = get_prefix(prov_subject.res)
252 related_to_label = "related to %s %s%s" % (GraphSet.labels[cur_short_name], cur_entity_prefix,
253 cur_entity_count)
254 related_to_short_label = "-> %s/%s%s" % (cur_short_name, cur_entity_prefix, cur_entity_count)
256 label = "%s %s %s [%s/%s %s]" % (self.labels[short_name], count, related_to_label, short_name, count,
257 related_to_short_label)
259 return cur_g, count, label
261 @staticmethod
262 def _set_ns(g: Graph) -> None:
263 g.namespace_manager.bind("prov", ProvEntity.PROV)
265 def _retrieve_last_snapshot(self, prov_subject: URIRef) -> Optional[URIRef]:
266 subj_short_name: str = get_short_name(prov_subject)
267 try:
268 subj_count: str = get_count(prov_subject)
269 if int(subj_count) <= 0:
270 raise ValueError('prov_subject is not a valid URIRef. Extracted count value should be a positive '
271 'non-zero integer number!')
272 except ValueError:
273 raise ValueError('prov_subject is not a valid URIRef. Unable to extract the count value!')
275 supplier_prefix = get_prefix(str(prov_subject))
277 if isinstance(self.counter_handler, SqliteCounterHandler):
278 last_snapshot_count: str = str(self.counter_handler.read_counter(prov_subject))
279 else:
280 last_snapshot_count: str = str(self.counter_handler.read_counter(subj_short_name, "se", int(subj_count), supplier_prefix=supplier_prefix))
282 if int(last_snapshot_count) <= 0:
283 return None
284 else:
285 return URIRef(str(prov_subject) + '/prov/se/' + last_snapshot_count)
287 def get_se(self) -> Tuple[SnapshotEntity]:
288 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, SnapshotEntity))