Coverage for rdflib_ocdm / prov / snapshot_entity.py: 94%
77 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-21 12:35 +0000
1#!/usr/bin/python
3# SPDX-FileCopyrightText: 2016 Silvio Peroni <essepuntato@gmail.com>
4# SPDX-FileCopyrightText: 2023-2025 Arcangelo Massari <arcangelo.massari@unibo.it>
5#
6# SPDX-License-Identifier: ISC
7from __future__ import annotations
9from typing import TYPE_CHECKING
11from oc_ocdm.decorators import accepts_only
12from rdflib import XSD
14from rdflib_ocdm.prov.prov_entity import ProvEntity
16if TYPE_CHECKING:
17 from typing import List, Optional
19 from rdflib import URIRef
22class SnapshotEntity(ProvEntity):
23 """Snapshot of entity metadata: a particular snapshot recording the
24 metadata associated with an individual entity at a particular date and time,
25 including the agent, such as a person, organisation or automated process that
26 created or modified the entity metadata."""
28 # HAS CREATION DATE
29 def get_generation_time(self) -> Optional[str]:
30 """
31 Getter method corresponding to the ``prov:generatedAtTime`` RDF predicate.
33 :return: The requested value if found, None otherwise
34 """
35 return self._get_literal(ProvEntity.iri_generated_at_time)
37 @accepts_only('literal')
38 def has_generation_time(self, string: str) -> None:
39 """
40 Setter method corresponding to the ``prov:generatedAtTime`` RDF predicate.
42 **WARNING: this is a functional property, hence any existing value will be overwritten!**
44 `The date on which a particular snapshot of a bibliographic entity's metadata was
45 created.`
47 :param string: The value that will be set as the object of the property related to this method. **It must
48 be a string compliant with the** ``xsd:dateTime`` **datatype.**
49 :type string: str
50 :raises TypeError: if the parameter is of the wrong type
51 :return: None
52 """
53 self.remove_generation_time()
54 self._create_literal(ProvEntity.iri_generated_at_time, string, XSD.dateTime)
56 def remove_generation_time(self) -> None:
57 """
58 Remover method corresponding to the ``prov:generatedAtTime`` RDF predicate.
60 :return: None
61 """
62 self.g.remove((self.res, ProvEntity.iri_generated_at_time, None))
64 # HAS INVALIDATION DATE
65 def get_invalidation_time(self) -> Optional[str]:
66 """
67 Getter method corresponding to the ``prov:invalidatedAtTime`` RDF predicate.
69 :return: The requested value if found, None otherwise
70 """
71 return self._get_literal(ProvEntity.iri_invalidated_at_time)
73 @accepts_only('literal')
74 def has_invalidation_time(self, string: str) -> None:
75 """
76 Setter method corresponding to the ``prov:invalidatedAtTime`` RDF predicate.
78 **WARNING: this is a functional property, hence any existing value will be overwritten!**
80 `The date on which a snapshot of a bibliographic entity's metadata was invalidated due
81 to an update (e.g. a correction, or the addition of some metadata that was not specified
82 in the previous snapshot), or due to a merger of the entity with another one.`
84 :param string: The value that will be set as the object of the property related to this method. **It must
85 be a string compliant with the** ``xsd:dateTime`` **datatype.**
86 :type string: str
87 :raises TypeError: if the parameter is of the wrong type
88 :return: None
89 """
90 self.remove_invalidation_time()
91 self._create_literal(ProvEntity.iri_invalidated_at_time, string, XSD.dateTime)
93 def remove_invalidation_time(self) -> None:
94 """
95 Remover method corresponding to the ``prov:invalidatedAtTime`` RDF predicate.
97 :return: None
98 """
99 self.g.remove((self.res, ProvEntity.iri_invalidated_at_time, None))
101 # IS SNAPSHOT OF
102 def get_is_snapshot_of(self) -> Optional[URIRef]:
103 """
104 Getter method corresponding to the ``prov:specializationOf`` RDF predicate.
106 :return: The requested value if found, None otherwise
107 """
108 uri: Optional[URIRef] = self._get_uri_reference(ProvEntity.iri_specialization_of)
109 return uri
111 def is_snapshot_of(self, en_res: URIRef) -> None:
112 """
113 Setter method corresponding to the ``prov:specializationOf`` RDF predicate.
115 **WARNING: this is a functional property, hence any existing value will be overwritten!**
117 `This property is used to link a snapshot of entity metadata to the bibliographic entity
118 to which the snapshot refers.`
120 :param en_res: The value that will be set as the object of the property related to this method
121 :type en_res: URIRef
122 :return: None
123 """
124 self.remove_is_snapshot_of()
125 self.g.add((self.res, ProvEntity.iri_specialization_of, en_res))
127 def remove_is_snapshot_of(self) -> None:
128 """
129 Remover method corresponding to the ``prov:specializationOf`` RDF predicate.
131 :return: None
132 """
133 self.g.remove((self.res, ProvEntity.iri_specialization_of, None))
135 # IS DERIVED FROM
136 def get_derives_from(self) -> List[ProvEntity]:
137 """
138 Getter method corresponding to the ``prov:wasDerivedFrom`` RDF predicate.
140 :return: A list containing the requested values if found, None otherwise
141 """
142 uri_list: List[URIRef] = self._get_multiple_uri_references(ProvEntity.iri_was_derived_from)
143 result: List[ProvEntity] = []
144 for uri in uri_list:
145 # TODO: what is the prov_subject of these snapshots?
146 prov_subj = uri.split('/prov/se/')[0]
147 result.append(self.g.add_se(prov_subj, uri))
148 return result
150 def derives_from(self, se_res: ProvEntity) -> None:
151 """
152 Setter method corresponding to the ``prov:wasDerivedFrom`` RDF predicate.
154 `This property is used to identify the immediately previous snapshot of entity metadata
155 associated with the same bibliographic entity.`
157 :param se_res: The value that will be set as the object of the property related to this method
158 :type se_res: ProvEntity
159 :raises TypeError: if the parameter is of the wrong type
160 :return: None
161 """
162 self.g.add((self.res, ProvEntity.iri_was_derived_from, se_res.res))
164 def remove_derives_from(self, se_res: ProvEntity = None) -> None:
165 """
166 Remover method corresponding to the ``prov:wasDerivedFrom`` RDF predicate.
168 **WARNING: this is a non-functional property, hence, if the parameter
169 is None, any existing value will be removed!**
171 :param se_res: If not None, the specific object value that will be removed from the property
172 related to this method (defaults to None)
173 :type se_res: SnapshotEntity
174 :raises TypeError: if the parameter is of the wrong type
175 :return: None
176 """
177 if se_res is not None:
178 self.g.remove((self.res, ProvEntity.iri_was_derived_from, se_res.res))
179 else:
180 self.g.remove((self.res, ProvEntity.iri_was_derived_from, None))
182 # HAS PRIMARY SOURCE
183 def get_primary_source(self) -> Optional[URIRef]:
184 """
185 Getter method corresponding to the ``prov:hadPrimarySource`` RDF predicate.
187 :return: The requested value if found, None otherwise
188 """
189 uri: Optional[URIRef] = self._get_uri_reference(ProvEntity.iri_had_primary_source)
190 return uri
192 @accepts_only('thing')
193 def has_primary_source(self, any_res: URIRef) -> None:
194 """
195 Setter method corresponding to the ``prov:hadPrimarySource`` RDF predicate.
197 **WARNING: this is a functional property, hence any existing value will be overwritten!**
199 `This property is used to identify the primary source from which the metadata
200 described in the snapshot are derived (e.g. Crossref, as the result of querying the
201 CrossRef API).`
203 :param any_res: The value that will be set as the object of the property related to this method
204 :type any_res: URIRef
205 :return: None
206 """
207 self.remove_primary_source()
208 self.g.add((self.res, ProvEntity.iri_had_primary_source, any_res))
210 def remove_primary_source(self) -> None:
211 """
212 Remover method corresponding to the ``prov:hadPrimarySource`` RDF predicate.
214 :return: None
215 """
216 self.g.remove((self.res, ProvEntity.iri_had_primary_source, None))
218 # HAS UPDATE ACTION
219 def get_update_action(self) -> Optional[str]:
220 """
221 Getter method corresponding to the ``oco:hasUpdateQuery`` RDF predicate.
223 :return: The requested value if found, None otherwise
224 """
225 return self._get_literal(ProvEntity.iri_has_update_query)
227 @accepts_only('literal')
228 def has_update_action(self, string: str) -> None:
229 """
230 Setter method corresponding to the ``oco:hasUpdateQuery`` RDF predicate.
232 **WARNING: this is a functional property, hence any existing value will be overwritten!**
234 `The UPDATE SPARQL query that specifies which data, associated to the bibliographic
235 entity in consideration, have been modified (e.g. for correcting a mistake) in the
236 current snapshot starting from those associated to the previous snapshot of the entity.`
238 :param string: The value that will be set as the object of the property related to this method
239 :type string: str
240 :return: None
241 """
242 self.remove_update_action()
243 self._create_literal(ProvEntity.iri_has_update_query, string)
245 def remove_update_action(self) -> None:
246 """
247 Remover method corresponding to the ``oco:hasUpdateQuery`` RDF predicate.
249 :return: None
250 """
251 self.g.remove((self.res, ProvEntity.iri_has_update_query, None))
253 # HAS DESCRIPTION
254 def get_description(self) -> Optional[str]:
255 """
256 Getter method corresponding to the ``dcterms:description`` RDF predicate.
258 :return: The requested value if found, None otherwise
259 """
260 return self._get_literal(ProvEntity.iri_description)
262 @accepts_only('literal')
263 def has_description(self, string: str) -> None:
264 """
265 Setter method corresponding to the ``dcterms:description`` RDF predicate.
267 **WARNING: this is a functional property, hence any existing value will be overwritten!**
269 `A textual description of the events that have resulted in the current snapshot (e.g. the
270 creation of the initial snapshot, the creation of a new snapshot following the
271 modification of the entity to which the metadata relate, or the creation of a new
272 snapshot following the merger with another entity of the entity to which the previous
273 snapshot related).`
275 :param string: The value that will be set as the object of the property related to this method
276 :type string: str
277 :return: None
278 """
279 self.remove_description()
280 self._create_literal(ProvEntity.iri_description, string)
282 def remove_description(self) -> None:
283 """
284 Remover method corresponding to the ``dcterms:description`` RDF predicate.
286 :return: None
287 """
288 self.g.remove((self.res, ProvEntity.iri_description, None))
290 # IS ATTRIBUTED TO
291 def get_resp_agent(self) -> Optional[URIRef]:
292 """
293 Getter method corresponding to the ``prov:wasAttributedTo`` RDF predicate.
295 :return: The requested value if found, None otherwise
296 """
297 uri: Optional[URIRef] = self._get_uri_reference(ProvEntity.iri_was_attributed_to)
298 return uri
300 @accepts_only('thing')
301 def has_resp_agent(self, se_agent: URIRef) -> None:
302 """
303 Setter method corresponding to the ``prov:wasAttributedTo`` RDF predicate.
305 **WARNING: this is a functional property, hence any existing value will be overwritten!**
307 `The agent responsible for the creation of the current entity snapshot.`
309 :param se_agent: The value that will be set as the object of the property related to this method
310 :type se_agent: URIRef
311 :return: None
312 """
313 self.remove_resp_agent()
314 self.g.add((self.res, ProvEntity.iri_was_attributed_to, se_agent))
316 def remove_resp_agent(self) -> None:
317 """
318 Remover method corresponding to the ``prov:wasAttributedTo`` RDF predicate.
320 :return: None
321 """
322 self.g.remove((self.res, ProvEntity.iri_was_attributed_to, None))