Coverage for oc_ocdm / prov / entities / snapshot_entity.py: 96%
79 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-28 18:52 +0000
1#!/usr/bin/python
3# SPDX-FileCopyrightText: 2021-2022 Simone Persiani <iosonopersia@gmail.com>
4#
5# SPDX-License-Identifier: ISC
7# -*- coding: utf-8 -*-
8from __future__ import annotations
10from typing import TYPE_CHECKING
12from rdflib import XSD
14from oc_ocdm.decorators import accepts_only
15from oc_ocdm.graph.graph_entity import GraphEntity
16from oc_ocdm.prov.prov_entity import ProvEntity
18if TYPE_CHECKING:
19 from typing import List, Optional
21 from rdflib import URIRef
24class SnapshotEntity(ProvEntity):
25 """Snapshot of entity metadata (short: se): a particular snapshot recording the
26 metadata associated with an individual entity (either a bibliographic entity or an
27 identifier) at a particular date and time, including the agent, such as a person,
28 organisation or automated process that created or modified the entity metadata."""
30 # HAS CREATION DATE
31 def get_generation_time(self) -> Optional[str]:
32 """
33 Getter method corresponding to the ``prov:generatedAtTime`` RDF predicate.
35 :return: The requested value if found, None otherwise
36 """
37 return self._get_literal(ProvEntity.iri_generated_at_time)
39 @accepts_only('literal')
40 def has_generation_time(self, string: str) -> None:
41 """
42 Setter method corresponding to the ``prov:generatedAtTime`` RDF predicate.
44 **WARNING: this is a functional property, hence any existing value will be overwritten!**
46 `The date on which a particular snapshot of a bibliographic entity's metadata was
47 created.`
49 :param string: The value that will be set as the object of the property related to this method. **It must
50 be a string compliant with the** ``xsd:dateTime`` **datatype.**
51 :type string: str
52 :raises TypeError: if the parameter is of the wrong type
53 :return: None
54 """
55 self.remove_generation_time()
56 self._create_literal(ProvEntity.iri_generated_at_time, string, XSD.dateTime)
58 def remove_generation_time(self) -> None:
59 """
60 Remover method corresponding to the ``prov:generatedAtTime`` RDF predicate.
62 :return: None
63 """
64 self.g.remove((self.res, ProvEntity.iri_generated_at_time, None))
66 # HAS INVALIDATION DATE
67 def get_invalidation_time(self) -> Optional[str]:
68 """
69 Getter method corresponding to the ``prov:invalidatedAtTime`` RDF predicate.
71 :return: The requested value if found, None otherwise
72 """
73 return self._get_literal(ProvEntity.iri_invalidated_at_time)
75 @accepts_only('literal')
76 def has_invalidation_time(self, string: str) -> None:
77 """
78 Setter method corresponding to the ``prov:invalidatedAtTime`` RDF predicate.
80 **WARNING: this is a functional property, hence any existing value will be overwritten!**
82 `The date on which a snapshot of a bibliographic entity's metadata was invalidated due
83 to an update (e.g. a correction, or the addition of some metadata that was not specified
84 in the previous snapshot), or due to a merger of the entity with another one.`
86 :param string: The value that will be set as the object of the property related to this method. **It must
87 be a string compliant with the** ``xsd:dateTime`` **datatype.**
88 :type string: str
89 :raises TypeError: if the parameter is of the wrong type
90 :return: None
91 """
92 self.remove_invalidation_time()
93 self._create_literal(ProvEntity.iri_invalidated_at_time, string, XSD.dateTime)
95 def remove_invalidation_time(self) -> None:
96 """
97 Remover method corresponding to the ``prov:invalidatedAtTime`` RDF predicate.
99 :return: None
100 """
101 self.g.remove((self.res, ProvEntity.iri_invalidated_at_time, None))
103 # IS SNAPSHOT OF
104 def get_is_snapshot_of(self) -> Optional[URIRef]:
105 """
106 Getter method corresponding to the ``prov:specializationOf`` RDF predicate.
108 :return: The requested value if found, None otherwise
109 """
110 uri: Optional[URIRef] = self._get_uri_reference(ProvEntity.iri_specialization_of)
111 return uri
113 def is_snapshot_of(self, en_res: GraphEntity) -> None:
114 """
115 Setter method corresponding to the ``prov:specializationOf`` RDF predicate.
117 **WARNING: this is a functional property, hence any existing value will be overwritten!**
119 `This property is used to link a snapshot of entity metadata to the bibliographic entity
120 to which the snapshot refers.`
122 :param en_res: The value that will be set as the object of the property related to this method
123 :type en_res: GraphEntity
124 :return: None
125 """
126 self.remove_is_snapshot_of()
127 self.g.add((self.res, ProvEntity.iri_specialization_of, en_res.res))
129 def remove_is_snapshot_of(self) -> None:
130 """
131 Remover method corresponding to the ``prov:specializationOf`` RDF predicate.
133 :return: None
134 """
135 self.g.remove((self.res, ProvEntity.iri_specialization_of, None))
137 # IS DERIVED FROM
138 def get_derives_from(self) -> List[ProvEntity]:
139 """
140 Getter method corresponding to the ``prov:wasDerivedFrom`` RDF predicate.
142 :return: A list containing the requested values if found, None otherwise
143 """
144 uri_list: List[URIRef] = self._get_multiple_uri_references(ProvEntity.iri_was_derived_from, 'se')
145 result: List[ProvEntity] = []
146 for uri in uri_list:
147 # TODO: what is the prov_subject of these snapshots?
148 result.append(self.p_set.add_se(self.prov_subject, uri))
149 return result
151 @accepts_only('se')
152 def derives_from(self, se_res: ProvEntity) -> None:
153 """
154 Setter method corresponding to the ``prov:wasDerivedFrom`` RDF predicate.
156 `This property is used to identify the immediately previous snapshot of entity metadata
157 associated with the same bibliographic entity.`
159 :param se_res: The value that will be set as the object of the property related to this method
160 :type se_res: ProvEntity
161 :raises TypeError: if the parameter is of the wrong type
162 :return: None
163 """
164 self.g.add((self.res, ProvEntity.iri_was_derived_from, se_res.res))
166 @accepts_only('se')
167 def remove_derives_from(self, se_res: Optional[ProvEntity] = None) -> None:
168 """
169 Remover method corresponding to the ``prov:wasDerivedFrom`` RDF predicate.
171 **WARNING: this is a non-functional property, hence, if the parameter
172 is None, any existing value will be removed!**
174 :param se_res: If not None, the specific object value that will be removed from the property
175 related to this method (defaults to None)
176 :type se_res: SnapshotEntity
177 :raises TypeError: if the parameter is of the wrong type
178 :return: None
179 """
180 if se_res is not None:
181 self.g.remove((self.res, ProvEntity.iri_was_derived_from, se_res.res))
182 else:
183 self.g.remove((self.res, ProvEntity.iri_was_derived_from, None))
185 # HAS PRIMARY SOURCE
186 def get_primary_source(self) -> Optional[URIRef]:
187 """
188 Getter method corresponding to the ``prov:hadPrimarySource`` RDF predicate.
190 :return: The requested value if found, None otherwise
191 """
192 uri: Optional[URIRef] = self._get_uri_reference(ProvEntity.iri_had_primary_source)
193 return uri
195 @accepts_only('thing')
196 def has_primary_source(self, any_res: URIRef) -> None:
197 """
198 Setter method corresponding to the ``prov:hadPrimarySource`` RDF predicate.
200 **WARNING: this is a functional property, hence any existing value will be overwritten!**
202 `This property is used to identify the primary source from which the metadata
203 described in the snapshot are derived (e.g. Crossref, as the result of querying the
204 CrossRef API).`
206 :param any_res: The value that will be set as the object of the property related to this method
207 :type any_res: URIRef
208 :return: None
209 """
210 self.remove_primary_source()
211 self.g.add((self.res, ProvEntity.iri_had_primary_source, any_res))
213 def remove_primary_source(self) -> None:
214 """
215 Remover method corresponding to the ``prov:hadPrimarySource`` RDF predicate.
217 :return: None
218 """
219 self.g.remove((self.res, ProvEntity.iri_had_primary_source, None))
221 # HAS UPDATE ACTION
222 def get_update_action(self) -> Optional[str]:
223 """
224 Getter method corresponding to the ``oco:hasUpdateQuery`` RDF predicate.
226 :return: The requested value if found, None otherwise
227 """
228 return self._get_literal(ProvEntity.iri_has_update_query)
230 @accepts_only('literal')
231 def has_update_action(self, string: str) -> None:
232 """
233 Setter method corresponding to the ``oco:hasUpdateQuery`` RDF predicate.
235 **WARNING: this is a functional property, hence any existing value will be overwritten!**
237 `The UPDATE SPARQL query that specifies which data, associated to the bibliographic
238 entity in consideration, have been modified (e.g. for correcting a mistake) in the
239 current snapshot starting from those associated to the previous snapshot of the entity.`
241 :param string: The value that will be set as the object of the property related to this method
242 :type string: str
243 :return: None
244 """
245 self.remove_update_action()
246 self._create_literal(ProvEntity.iri_has_update_query, string)
248 def remove_update_action(self) -> None:
249 """
250 Remover method corresponding to the ``oco:hasUpdateQuery`` RDF predicate.
252 :return: None
253 """
254 self.g.remove((self.res, ProvEntity.iri_has_update_query, None))
256 # HAS DESCRIPTION
257 def get_description(self) -> Optional[str]:
258 """
259 Getter method corresponding to the ``dcterms:description`` RDF predicate.
261 :return: The requested value if found, None otherwise
262 """
263 return self._get_literal(ProvEntity.iri_description)
265 @accepts_only('literal')
266 def has_description(self, string: str) -> None:
267 """
268 Setter method corresponding to the ``dcterms:description`` RDF predicate.
270 **WARNING: this is a functional property, hence any existing value will be overwritten!**
272 `A textual description of the events that have resulted in the current snapshot (e.g. the
273 creation of the initial snapshot, the creation of a new snapshot following the
274 modification of the entity to which the metadata relate, or the creation of a new
275 snapshot following the merger with another entity of the entity to which the previous
276 snapshot related).`
278 :param string: The value that will be set as the object of the property related to this method
279 :type string: str
280 :return: None
281 """
282 self.remove_description()
283 self._create_literal(ProvEntity.iri_description, string)
285 def remove_description(self) -> None:
286 """
287 Remover method corresponding to the ``dcterms:description`` RDF predicate.
289 :return: None
290 """
291 self.g.remove((self.res, ProvEntity.iri_description, None))
293 # IS ATTRIBUTED TO
294 def get_resp_agent(self) -> Optional[URIRef]:
295 """
296 Getter method corresponding to the ``prov:wasAttributedTo`` RDF predicate.
298 :return: The requested value if found, None otherwise
299 """
300 uri: Optional[URIRef] = self._get_uri_reference(ProvEntity.iri_was_attributed_to)
301 return uri
303 @accepts_only('thing')
304 def has_resp_agent(self, se_agent: URIRef) -> None:
305 """
306 Setter method corresponding to the ``prov:wasAttributedTo`` RDF predicate.
308 **WARNING: this is a functional property, hence any existing value will be overwritten!**
310 `The agent responsible for the creation of the current entity snapshot.`
312 :param se_agent: The value that will be set as the object of the property related to this method
313 :type se_agent: URIRef
314 :return: None
315 """
316 self.remove_resp_agent()
317 self.g.add((self.res, ProvEntity.iri_was_attributed_to, se_agent))
319 def remove_resp_agent(self) -> None:
320 """
321 Remover method corresponding to the ``prov:wasAttributedTo`` RDF predicate.
323 :return: None
324 """
325 self.g.remove((self.res, ProvEntity.iri_was_attributed_to, None))