Coverage for rdflib_ocdm / prov / provenance.py: 95%
124 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-30 21:23 +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
11if TYPE_CHECKING:
12 from typing import Dict, List, Optional
14 from rdflib_ocdm.ocdm_graph import OCDMGraphCommons
16from collections import OrderedDict
17from datetime import datetime, timezone
19from rdflib import Dataset, URIRef
21from rdflib_ocdm.counter_handler.counter_handler import CounterHandler
22from rdflib_ocdm.counter_handler.in_memory_counter_handler import InMemoryCounterHandler
23from rdflib_ocdm.prov.prov_entity import ProvEntity
24from rdflib_ocdm.prov.snapshot_entity import SnapshotEntity
25from rdflib_ocdm.query_utils import get_update_query
26from rdflib_ocdm.support import get_prov_count
29class OCDMProvenance(Dataset):
30 def __init__(
31 self,
32 prov_subj_graph: OCDMGraphCommons,
33 counter_handler: CounterHandler | None = None,
34 ):
35 Dataset.__init__(self)
36 self.prov_g = prov_subj_graph
37 # The following variable maps a URIRef with the related provenance entity
38 self.res_to_entity: Dict[str, ProvEntity] = dict()
39 self.all_entities = set()
40 if counter_handler is None:
41 counter_handler = InMemoryCounterHandler()
42 self.counter_handler = counter_handler
44 def generate_provenance(self, c_time: float | None = None) -> None:
45 if c_time is None:
46 cur_time: str = (
47 datetime.now(tz=timezone.utc).replace(microsecond=0).isoformat(sep="T")
48 )
49 else:
50 cur_time: str = (
51 datetime.fromtimestamp(c_time, tz=timezone.utc)
52 .replace(microsecond=0)
53 .isoformat(sep="T")
54 )
55 merge_index = self.prov_g.merge_index
56 prov_g_subjects = OrderedDict(
57 sorted(
58 self.prov_g.entity_index.items(),
59 key=lambda x: not x[1]["to_be_deleted"],
60 reverse=True,
61 )
62 )
63 for cur_subj, cur_subj_metadata in prov_g_subjects.items():
64 last_snapshot_res: Optional[URIRef] = self._retrieve_last_snapshot(cur_subj)
65 if cur_subj_metadata["to_be_deleted"]:
66 update_query: str = get_update_query(self.prov_g, cur_subj)[0]
67 # DELETION SNAPSHOT
68 last_snapshot: SnapshotEntity = self.add_se(
69 prov_subject=cur_subj, res=last_snapshot_res
70 )
71 last_snapshot.has_invalidation_time(cur_time)
73 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
74 cur_snapshot.derives_from(last_snapshot)
75 cur_snapshot.has_invalidation_time(cur_time)
76 cur_snapshot.has_description(
77 f"The entity '{str(cur_subj)}' has been deleted."
78 )
79 cur_snapshot.has_update_action(update_query)
80 elif cur_subj_metadata["is_restored"]:
81 # RESTORATION SNAPSHOT
82 last_snapshot: SnapshotEntity = self.add_se(
83 prov_subject=cur_subj, res=last_snapshot_res
84 )
85 # Non settiamo l'invalidation time per il precedente
86 # snapshot in caso di restore
88 cur_snapshot: SnapshotEntity = self._create_snapshot(cur_subj, cur_time)
89 cur_snapshot.derives_from(last_snapshot)
90 cur_snapshot.has_description(
91 f"The entity '{str(cur_subj)}' has been restored."
92 )
94 update_query: str = get_update_query(self.prov_g, cur_subj)[0]
95 if update_query: 95 ↛ 63line 95 didn't jump to line 63 because the condition on line 95 was always true
96 cur_snapshot.has_update_action(update_query)
97 else:
98 if last_snapshot_res is None:
99 # CREATION SNAPSHOT
100 cur_snapshot: SnapshotEntity = self._create_snapshot(
101 cur_subj, cur_time
102 )
103 cur_snapshot.has_description(
104 f"The entity '{str(cur_subj)}' has been created."
105 )
106 else:
107 update_query = get_update_query(self.prov_g, cur_subj)[0]
108 cur_subj_merge_index = {
109 k: v for k, v in merge_index.items() if k == cur_subj
110 }
111 snapshots_list = self._get_snapshots_from_merge_list(
112 cur_subj_merge_index
113 )
114 if update_query and len(snapshots_list) == 0:
115 # MODIFICATION SNAPSHOT
116 last_snapshot: SnapshotEntity = self.add_se(
117 prov_subject=cur_subj, res=last_snapshot_res
118 )
119 last_snapshot.has_invalidation_time(cur_time)
120 cur_snapshot: SnapshotEntity = self._create_snapshot(
121 cur_subj, cur_time
122 )
123 cur_snapshot.derives_from(last_snapshot)
124 cur_snapshot.has_description(
125 f"The entity '{str(cur_subj)}' was modified."
126 )
127 cur_snapshot.has_update_action(update_query)
128 elif len(snapshots_list) > 0:
129 # MERGE SNAPSHOT
130 last_snapshot: SnapshotEntity = self.add_se(
131 prov_subject=cur_subj, res=last_snapshot_res
132 )
133 last_snapshot.has_invalidation_time(cur_time)
134 cur_snapshot: SnapshotEntity = self._create_snapshot(
135 cur_subj, cur_time
136 )
137 cur_snapshot.derives_from(last_snapshot)
138 for snapshot in snapshots_list:
139 cur_snapshot.derives_from(snapshot)
140 if update_query: 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true
141 cur_snapshot.has_update_action(update_query)
142 cur_snapshot.has_description(
143 self._get_merge_description(cur_subj, snapshots_list)
144 )
146 @staticmethod
147 def _get_merge_description(
148 cur_subj: URIRef, snapshots_list: List[SnapshotEntity]
149 ) -> str:
150 merge_description: str = f"The entity '{str(cur_subj)}' was merged"
151 is_first: bool = True
152 for snapshot in snapshots_list:
153 if is_first: 153 ↛ 157line 153 didn't jump to line 157 because the condition on line 153 was always true
154 merge_description += f" with '{snapshot.prov_subject}'"
155 is_first = False
156 else:
157 merge_description += f", '{snapshot.prov_subject}'"
158 merge_description += "."
159 return merge_description
161 def _retrieve_last_snapshot(self, prov_subject: URIRef) -> URIRef | None:
162 last_snapshot_count: str = str(
163 self.counter_handler.read_counter(str(prov_subject))
164 )
165 if int(last_snapshot_count) <= 0:
166 return None
167 else:
168 return URIRef(str(prov_subject) + "/prov/se/" + last_snapshot_count)
170 def _create_snapshot(self, cur_subj: URIRef, cur_time: str) -> SnapshotEntity:
171 new_snapshot: SnapshotEntity = self.add_se(prov_subject=cur_subj)
172 new_snapshot.is_snapshot_of(cur_subj)
173 new_snapshot.has_generation_time(cur_time)
174 source = self.prov_g.entity_index[cur_subj]["source"]
175 resp_agent = self.prov_g.entity_index[cur_subj]["resp_agent"]
176 if source is not None:
177 new_snapshot.has_primary_source(URIRef(source))
178 if resp_agent is not None:
179 new_snapshot.has_resp_agent(URIRef(resp_agent))
180 return new_snapshot
182 def _get_snapshots_from_merge_list(
183 self, cur_subj_merge_index: dict
184 ) -> List[SnapshotEntity]:
185 snapshots_list: List[SnapshotEntity] = []
186 for _, merge_entities in cur_subj_merge_index.items():
187 for merge_entity in merge_entities:
188 last_entity_snapshot_res: Optional[URIRef] = (
189 self._retrieve_last_snapshot(merge_entity)
190 )
191 if last_entity_snapshot_res is not None: 191 ↛ 187line 191 didn't jump to line 187 because the condition on line 191 was always true
192 snapshots_list.append(
193 self.add_se(
194 prov_subject=merge_entity, res=last_entity_snapshot_res
195 )
196 )
197 return snapshots_list
199 def add_se(self, prov_subject: URIRef, res: URIRef | None = None) -> SnapshotEntity:
200 if res is not None and str(res) in self.res_to_entity:
201 entity = self.res_to_entity[str(res)]
202 assert isinstance(entity, SnapshotEntity)
203 return entity
204 count = self._add_prov(str(prov_subject), res)
205 se = SnapshotEntity(str(prov_subject), self, count)
206 return se
208 def _add_prov(self, prov_subject: str, res: URIRef | None) -> str:
209 if res is not None:
210 prov_count = get_prov_count(res)
211 assert prov_count is not None
212 res_count: int = int(prov_count)
213 if res_count > self.counter_handler.read_counter(prov_subject): 213 ↛ 214line 213 didn't jump to line 214 because the condition on line 213 was never true
214 self.counter_handler.set_counter(res_count, prov_subject)
215 return str(res_count)
216 return str(self.counter_handler.increment_counter(prov_subject))
218 def get_entity(self, res: str) -> ProvEntity | None:
219 if res in self.res_to_entity:
220 return self.res_to_entity[res]
221 return None