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

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 

17 

18from typing import TYPE_CHECKING 

19 

20from rdflib import XSD 

21 

22from oc_ocdm.decorators import accepts_only 

23from oc_ocdm.graph.graph_entity import GraphEntity 

24from oc_ocdm.prov.prov_entity import ProvEntity 

25 

26if TYPE_CHECKING: 

27 from typing import List, Optional 

28 

29 from rdflib import URIRef 

30 

31 

32class SnapshotEntity(ProvEntity): 

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

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

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

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

37 

38 # HAS CREATION DATE 

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

40 """ 

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

42 

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

44 """ 

45 return self._get_literal(ProvEntity.iri_generated_at_time) 

46 

47 @accepts_only('literal') 

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

49 """ 

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

51 

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

53 

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

55 created.` 

56 

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

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

59 :type string: str 

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

61 :return: None 

62 """ 

63 self.remove_generation_time() 

64 self._create_literal(ProvEntity.iri_generated_at_time, string, XSD.dateTime) 

65 

66 def remove_generation_time(self) -> None: 

67 """ 

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

69 

70 :return: None 

71 """ 

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

73 

74 # HAS INVALIDATION DATE 

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

76 """ 

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

78 

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

80 """ 

81 return self._get_literal(ProvEntity.iri_invalidated_at_time) 

82 

83 @accepts_only('literal') 

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

85 """ 

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

87 

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

89 

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

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

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

93 

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

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

96 :type string: str 

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

98 :return: None 

99 """ 

100 self.remove_invalidation_time() 

101 self._create_literal(ProvEntity.iri_invalidated_at_time, string, XSD.dateTime) 

102 

103 def remove_invalidation_time(self) -> None: 

104 """ 

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

106 

107 :return: None 

108 """ 

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

110 

111 # IS SNAPSHOT OF 

112 def get_is_snapshot_of(self) -> Optional[URIRef]: 

113 """ 

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

115 

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

117 """ 

118 uri: Optional[URIRef] = self._get_uri_reference(ProvEntity.iri_specialization_of) 

119 return uri 

120 

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

122 """ 

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

124 

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

126 

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

128 to which the snapshot refers.` 

129 

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

131 :type en_res: GraphEntity 

132 :return: None 

133 """ 

134 self.remove_is_snapshot_of() 

135 self.g.add((self.res, ProvEntity.iri_specialization_of, en_res.res)) 

136 

137 def remove_is_snapshot_of(self) -> None: 

138 """ 

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

140 

141 :return: None 

142 """ 

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

144 

145 # IS DERIVED FROM 

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

147 """ 

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

149 

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

151 """ 

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

153 result: List[ProvEntity] = [] 

154 for uri in uri_list: 

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

156 result.append(self.p_set.add_se(None, uri)) 

157 return result 

158 

159 @accepts_only('se') 

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

161 """ 

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

163 

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

165 associated with the same bibliographic entity.` 

166 

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

168 :type se_res: ProvEntity 

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

170 :return: None 

171 """ 

172 self.g.add((self.res, ProvEntity.iri_was_derived_from, se_res.res)) 

173 

174 @accepts_only('se') 

175 def remove_derives_from(self, se_res: ProvEntity = None) -> None: 

176 """ 

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

178 

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

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

181 

182 :param se_res: If not None, the specific object value that will be removed from the property 

183 related to this method (defaults to None) 

184 :type se_res: SnapshotEntity 

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

186 :return: None 

187 """ 

188 if se_res is not None: 

189 self.g.remove((self.res, ProvEntity.iri_was_derived_from, se_res.res)) 

190 else: 

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

192 

193 # HAS PRIMARY SOURCE 

194 def get_primary_source(self) -> Optional[URIRef]: 

195 """ 

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

197 

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

199 """ 

200 uri: Optional[URIRef] = self._get_uri_reference(ProvEntity.iri_had_primary_source) 

201 return uri 

202 

203 @accepts_only('thing') 

204 def has_primary_source(self, any_res: URIRef) -> None: 

205 """ 

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

207 

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

209 

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

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

212 CrossRef API).` 

213 

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

215 :type any_res: URIRef 

216 :return: None 

217 """ 

218 self.remove_primary_source() 

219 self.g.add((self.res, ProvEntity.iri_had_primary_source, any_res)) 

220 

221 def remove_primary_source(self) -> None: 

222 """ 

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

224 

225 :return: None 

226 """ 

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

228 

229 # HAS UPDATE ACTION 

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

231 """ 

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

233 

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

235 """ 

236 return self._get_literal(ProvEntity.iri_has_update_query) 

237 

238 @accepts_only('literal') 

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

240 """ 

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

242 

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

244 

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

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

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

248 

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

250 :type string: str 

251 :return: None 

252 """ 

253 self.remove_update_action() 

254 self._create_literal(ProvEntity.iri_has_update_query, string) 

255 

256 def remove_update_action(self) -> None: 

257 """ 

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

259 

260 :return: None 

261 """ 

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

263 

264 # HAS DESCRIPTION 

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

266 """ 

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

268 

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

270 """ 

271 return self._get_literal(ProvEntity.iri_description) 

272 

273 @accepts_only('literal') 

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

275 """ 

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

277 

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

279 

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

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

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

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

284 snapshot related).` 

285 

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

287 :type string: str 

288 :return: None 

289 """ 

290 self.remove_description() 

291 self._create_literal(ProvEntity.iri_description, string) 

292 

293 def remove_description(self) -> None: 

294 """ 

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

296 

297 :return: None 

298 """ 

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

300 

301 # IS ATTRIBUTED TO 

302 def get_resp_agent(self) -> Optional[URIRef]: 

303 """ 

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

305 

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

307 """ 

308 uri: Optional[URIRef] = self._get_uri_reference(ProvEntity.iri_was_attributed_to) 

309 return uri 

310 

311 @accepts_only('thing') 

312 def has_resp_agent(self, se_agent: URIRef) -> None: 

313 """ 

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

315 

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

317 

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

319 

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

321 :type se_agent: URIRef 

322 :return: None 

323 """ 

324 self.remove_resp_agent() 

325 self.g.add((self.res, ProvEntity.iri_was_attributed_to, se_agent)) 

326 

327 def remove_resp_agent(self) -> None: 

328 """ 

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

330 

331 :return: None 

332 """ 

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