Coverage for oc_ocdm / prov / entities / snapshot_entity.py: 96%

1#!/usr/bin/python 

2 

3# SPDX-FileCopyrightText: 2021-2022 Simone Persiani <iosonopersia@gmail.com> 

4# 

5# SPDX-License-Identifier: ISC 

6 

7# -*- coding: utf-8 -*- 

8from __future__ import annotations 

9 

10from typing import TYPE_CHECKING 

11 

12from triplelite import RDFTerm 

13 

14from oc_ocdm.constants import XSD_DATETIME 

15from oc_ocdm.decorators import accepts_only 

16from oc_ocdm.graph.graph_entity import GraphEntity 

17from oc_ocdm.prov.prov_entity import ProvEntity 

18 

19if TYPE_CHECKING: 

20 from typing import List, Optional 

21 

22 

23class SnapshotEntity(ProvEntity): 

24 """Snapshot of entity metadata (short: se): a particular snapshot recording the 

25 metadata associated with an individual entity (either a bibliographic entity or an 

26 identifier) at a particular date and time, including the agent, such as a person, 

27 organisation or automated process that created or modified the entity metadata.""" 

28 

29 # HAS CREATION DATE 

30 def get_generation_time(self) -> Optional[str]: 

31 """ 

32 Getter method corresponding to the ``prov:generatedAtTime`` RDF predicate. 

33 

34 :return: The requested value if found, None otherwise 

35 """ 

36 return self._get_literal(ProvEntity.iri_generated_at_time) 

37 

38 def has_generation_time(self, string: str) -> None: 

39 """ 

40 Setter method corresponding to the ``prov:generatedAtTime`` RDF predicate. 

41 

42 **WARNING: this is a functional property, hence any existing value will be overwritten!** 

43 

44 `The date on which a particular snapshot of a bibliographic entity's metadata was 

45 created.` 

46 

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) 

55 

56 def remove_generation_time(self) -> None: 

57 """ 

58 Remover method corresponding to the ``prov:generatedAtTime`` RDF predicate. 

59 

60 :return: None 

61 """ 

62 self.g.remove((self.res, ProvEntity.iri_generated_at_time, None)) 

63 

64 # HAS INVALIDATION DATE 

65 def get_invalidation_time(self) -> Optional[str]: 

66 """ 

67 Getter method corresponding to the ``prov:invalidatedAtTime`` RDF predicate. 

68 

69 :return: The requested value if found, None otherwise 

70 """ 

71 return self._get_literal(ProvEntity.iri_invalidated_at_time) 

72 

73 def has_invalidation_time(self, string: str) -> None: 

74 """ 

75 Setter method corresponding to the ``prov:invalidatedAtTime`` RDF predicate. 

76 

77 **WARNING: this is a functional property, hence any existing value will be overwritten!** 

78 

79 `The date on which a snapshot of a bibliographic entity's metadata was invalidated due 

80 to an update (e.g. a correction, or the addition of some metadata that was not specified 

81 in the previous snapshot), or due to a merger of the entity with another one.` 

82 

83 :param string: The value that will be set as the object of the property related to this method. **It must 

84 be a string compliant with the** ``xsd:dateTime`` **datatype.** 

85 :type string: str 

86 :raises TypeError: if the parameter is of the wrong type 

87 :return: None 

88 """ 

89 self.remove_invalidation_time() 

90 self._create_literal(ProvEntity.iri_invalidated_at_time, string, XSD_DATETIME) 

91 

92 def remove_invalidation_time(self) -> None: 

93 """ 

94 Remover method corresponding to the ``prov:invalidatedAtTime`` RDF predicate. 

95 

96 :return: None 

97 """ 

98 self.g.remove((self.res, ProvEntity.iri_invalidated_at_time, None)) 

99 

100 # IS SNAPSHOT OF 

101 def get_is_snapshot_of(self) -> Optional[str]: 

102 """ 

103 Getter method corresponding to the ``prov:specializationOf`` RDF predicate. 

104 

105 :return: The requested value if found, None otherwise 

106 """ 

107 uri: Optional[str] = self._get_uri_reference(ProvEntity.iri_specialization_of) 

108 return uri 

109 

110 def is_snapshot_of(self, en_res: GraphEntity) -> None: 

111 """ 

112 Setter method corresponding to the ``prov:specializationOf`` RDF predicate. 

113 

114 **WARNING: this is a functional property, hence any existing value will be overwritten!** 

115 

116 `This property is used to link a snapshot of entity metadata to the bibliographic entity 

117 to which the snapshot refers.` 

118 

119 :param en_res: The value that will be set as the object of the property related to this method 

120 :type en_res: GraphEntity 

121 :return: None 

122 """ 

123 self.remove_is_snapshot_of() 

124 self.g.add((self.res, ProvEntity.iri_specialization_of, RDFTerm("uri", str(en_res.res)))) 

125 

126 def remove_is_snapshot_of(self) -> None: 

127 """ 

128 Remover method corresponding to the ``prov:specializationOf`` RDF predicate. 

129 

130 :return: None 

131 """ 

132 self.g.remove((self.res, ProvEntity.iri_specialization_of, None)) 

133 

134 # IS DERIVED FROM 

135 def get_derives_from(self) -> List[ProvEntity]: 

136 """ 

137 Getter method corresponding to the ``prov:wasDerivedFrom`` RDF predicate. 

138 

139 :return: A list containing the requested values if found, None otherwise 

140 """ 

141 uri_list: List[str] = self._get_multiple_uri_references(ProvEntity.iri_was_derived_from, 'se') 

142 result: List[ProvEntity] = [] 

143 for uri in uri_list: 

144 # TODO: what is the prov_subject of these snapshots? 

145 result.append(self.p_set.add_se(self.prov_subject, uri)) 

146 return result 

147 

148 @accepts_only('se') 

149 def derives_from(self, se_res: ProvEntity) -> None: 

150 """ 

151 Setter method corresponding to the ``prov:wasDerivedFrom`` RDF predicate. 

152 

153 `This property is used to identify the immediately previous snapshot of entity metadata 

154 associated with the same bibliographic entity.` 

155 

156 :param se_res: The value that will be set as the object of the property related to this method 

157 :type se_res: ProvEntity 

158 :raises TypeError: if the parameter is of the wrong type 

159 :return: None 

160 """ 

161 self.g.add((self.res, ProvEntity.iri_was_derived_from, RDFTerm("uri", str(se_res.res)))) 

162 

163 @accepts_only('se') 

164 def remove_derives_from(self, se_res: Optional[ProvEntity] = None) -> None: 

165 """ 

166 Remover method corresponding to the ``prov:wasDerivedFrom`` RDF predicate. 

167 

168 **WARNING: this is a non-functional property, hence, if the parameter 

169 is None, any existing value will be removed!** 

170 

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, RDFTerm("uri", str(se_res.res)))) 

179 else: 

180 self.g.remove((self.res, ProvEntity.iri_was_derived_from, None)) 

181 

182 # HAS PRIMARY SOURCE 

183 def get_primary_source(self) -> Optional[str]: 

184 """ 

185 Getter method corresponding to the ``prov:hadPrimarySource`` RDF predicate. 

186 

187 :return: The requested value if found, None otherwise 

188 """ 

189 uri: Optional[str] = self._get_uri_reference(ProvEntity.iri_had_primary_source) 

190 return uri 

191 

192 def has_primary_source(self, any_res: str) -> None: 

193 """ 

194 Setter method corresponding to the ``prov:hadPrimarySource`` RDF predicate. 

195 

196 **WARNING: this is a functional property, hence any existing value will be overwritten!** 

197 

198 `This property is used to identify the primary source from which the metadata 

199 described in the snapshot are derived (e.g. Crossref, as the result of querying the 

200 CrossRef API).` 

201 

202 :param any_res: The value that will be set as the object of the property related to this method 

203 :type any_res: URIRef 

204 :return: None 

205 """ 

206 self.remove_primary_source() 

207 self.g.add((self.res, ProvEntity.iri_had_primary_source, RDFTerm("uri", str(any_res)))) 

208 

209 def remove_primary_source(self) -> None: 

210 """ 

211 Remover method corresponding to the ``prov:hadPrimarySource`` RDF predicate. 

212 

213 :return: None 

214 """ 

215 self.g.remove((self.res, ProvEntity.iri_had_primary_source, None)) 

216 

217 # HAS UPDATE ACTION 

218 def get_update_action(self) -> Optional[str]: 

219 """ 

220 Getter method corresponding to the ``oco:hasUpdateQuery`` RDF predicate. 

221 

222 :return: The requested value if found, None otherwise 

223 """ 

224 return self._get_literal(ProvEntity.iri_has_update_query) 

225 

226 def has_update_action(self, string: str) -> None: 

227 """ 

228 Setter method corresponding to the ``oco:hasUpdateQuery`` RDF predicate. 

229 

230 **WARNING: this is a functional property, hence any existing value will be overwritten!** 

231 

232 `The UPDATE SPARQL query that specifies which data, associated to the bibliographic 

233 entity in consideration, have been modified (e.g. for correcting a mistake) in the 

234 current snapshot starting from those associated to the previous snapshot of the entity.` 

235 

236 :param string: The value that will be set as the object of the property related to this method 

237 :type string: str 

238 :return: None 

239 """ 

240 self.remove_update_action() 

241 self._create_literal(ProvEntity.iri_has_update_query, string) 

242 

243 def remove_update_action(self) -> None: 

244 """ 

245 Remover method corresponding to the ``oco:hasUpdateQuery`` RDF predicate. 

246 

247 :return: None 

248 """ 

249 self.g.remove((self.res, ProvEntity.iri_has_update_query, None)) 

250 

251 # HAS DESCRIPTION 

252 def get_description(self) -> Optional[str]: 

253 """ 

254 Getter method corresponding to the ``dcterms:description`` RDF predicate. 

255 

256 :return: The requested value if found, None otherwise 

257 """ 

258 return self._get_literal(ProvEntity.iri_description) 

259 

260 def has_description(self, string: str) -> None: 

261 """ 

262 Setter method corresponding to the ``dcterms:description`` RDF predicate. 

263 

264 **WARNING: this is a functional property, hence any existing value will be overwritten!** 

265 

266 `A textual description of the events that have resulted in the current snapshot (e.g. the 

267 creation of the initial snapshot, the creation of a new snapshot following the 

268 modification of the entity to which the metadata relate, or the creation of a new 

269 snapshot following the merger with another entity of the entity to which the previous 

270 snapshot related).` 

271 

272 :param string: The value that will be set as the object of the property related to this method 

273 :type string: str 

274 :return: None 

275 """ 

276 self.remove_description() 

277 self._create_literal(ProvEntity.iri_description, string) 

278 

279 def remove_description(self) -> None: 

280 """ 

281 Remover method corresponding to the ``dcterms:description`` RDF predicate. 

282 

283 :return: None 

284 """ 

285 self.g.remove((self.res, ProvEntity.iri_description, None)) 

286 

287 # IS ATTRIBUTED TO 

288 def get_resp_agent(self) -> Optional[str]: 

289 """ 

290 Getter method corresponding to the ``prov:wasAttributedTo`` RDF predicate. 

291 

292 :return: The requested value if found, None otherwise 

293 """ 

294 uri: Optional[str] = self._get_uri_reference(ProvEntity.iri_was_attributed_to) 

295 return uri 

296 

297 def has_resp_agent(self, se_agent: str) -> None: 

298 """ 

299 Setter method corresponding to the ``prov:wasAttributedTo`` RDF predicate. 

300 

301 **WARNING: this is a functional property, hence any existing value will be overwritten!** 

302 

303 `The agent responsible for the creation of the current entity snapshot.` 

304 

305 :param se_agent: The value that will be set as the object of the property related to this method 

306 :type se_agent: URIRef 

307 :return: None 

308 """ 

309 self.remove_resp_agent() 

310 self.g.add((self.res, ProvEntity.iri_was_attributed_to, RDFTerm("uri", str(se_agent)))) 

311 

312 def remove_resp_agent(self) -> None: 

313 """ 

314 Remover method corresponding to the ``prov:wasAttributedTo`` RDF predicate. 

315 

316 :return: None 

317 """ 

318 self.g.remove((self.res, ProvEntity.iri_was_attributed_to, None))