Coverage for rdflib_ocdm/prov/snapshot_entity.py: 91%

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 oc_ocdm.decorators import accepts_only 

21from rdflib import XSD 

22 

23from rdflib_ocdm.prov.prov_entity import ProvEntity 

24 

25if TYPE_CHECKING: 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was never true

26 from typing import List, Optional 

27 

28 from rdflib import URIRef 

29 

30 

31class SnapshotEntity(ProvEntity): 

32 """Snapshot of entity metadata: a particular snapshot recording the 

33 metadata associated with an individual entity at a particular date and time,  

34 including the agent, such as a person, organisation or automated process that 

35 created or modified the entity metadata.""" 

36 

37 # HAS CREATION DATE 

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

39 """ 

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

41 

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

43 """ 

44 return self._get_literal(ProvEntity.iri_generated_at_time) 

45 

46 @accepts_only('literal') 

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

48 """ 

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

50 

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

52 

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

54 created.` 

55 

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

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

58 :type string: str 

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

60 :return: None 

61 """ 

62 self.remove_generation_time() 

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

64 

65 def remove_generation_time(self) -> None: 

66 """ 

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

68 

69 :return: None 

70 """ 

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

72 

73 # HAS INVALIDATION DATE 

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

75 """ 

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

77 

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

79 """ 

80 return self._get_literal(ProvEntity.iri_invalidated_at_time) 

81 

82 @accepts_only('literal') 

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

84 """ 

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

86 

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

88 

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

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

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

92 

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

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

95 :type string: str 

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

97 :return: None 

98 """ 

99 self.remove_invalidation_time() 

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

101 

102 def remove_invalidation_time(self) -> None: 

103 """ 

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

105 

106 :return: None 

107 """ 

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

109 

110 # IS SNAPSHOT OF 

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

112 """ 

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

114 

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

116 """ 

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

118 return uri 

119 

120 def is_snapshot_of(self, en_res: URIRef) -> None: 

121 """ 

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

123 

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

125 

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

127 to which the snapshot refers.` 

128 

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

130 :type en_res: URIRef 

131 :return: None 

132 """ 

133 self.remove_is_snapshot_of() 

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

135 

136 def remove_is_snapshot_of(self) -> None: 

137 """ 

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

139 

140 :return: None 

141 """ 

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

143 

144 # IS DERIVED FROM 

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

146 """ 

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

148 

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

150 """ 

151 uri_list: List[URIRef] = self._get_multiple_uri_references(ProvEntity.iri_was_derived_from) 

152 result: List[ProvEntity] = [] 

153 for uri in uri_list: 

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

155 prov_subj = uri.split('/prov/se/')[0] 

156 result.append(self.g.add_se(prov_subj, uri)) 

157 return result 

158 

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

160 """ 

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

162 

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

164 associated with the same bibliographic entity.` 

165 

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

167 :type se_res: ProvEntity 

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

169 :return: None 

170 """ 

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

172 

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

174 """ 

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

176 

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

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

179 

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

181 related to this method (defaults to None) 

182 :type se_res: SnapshotEntity 

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

184 :return: None 

185 """ 

186 if se_res is not None: 

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

188 else: 

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

190 

191 # HAS PRIMARY SOURCE 

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

193 """ 

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

195 

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

197 """ 

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

199 return uri 

200 

201 @accepts_only('thing') 

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

203 """ 

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

205 

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

207 

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

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

210 CrossRef API).` 

211 

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

213 :type any_res: URIRef 

214 :return: None 

215 """ 

216 self.remove_primary_source() 

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

218 

219 def remove_primary_source(self) -> None: 

220 """ 

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

222 

223 :return: None 

224 """ 

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

226 

227 # HAS UPDATE ACTION 

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

229 """ 

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

231 

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

233 """ 

234 return self._get_literal(ProvEntity.iri_has_update_query) 

235 

236 @accepts_only('literal') 

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

238 """ 

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

240 

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

242 

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

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

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

246 

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

248 :type string: str 

249 :return: None 

250 """ 

251 self.remove_update_action() 

252 self._create_literal(ProvEntity.iri_has_update_query, string) 

253 

254 def remove_update_action(self) -> None: 

255 """ 

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

257 

258 :return: None 

259 """ 

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

261 

262 # HAS DESCRIPTION 

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

264 """ 

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

266 

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

268 """ 

269 return self._get_literal(ProvEntity.iri_description) 

270 

271 @accepts_only('literal') 

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

273 """ 

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

275 

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

277 

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

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

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

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

282 snapshot related).` 

283 

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

285 :type string: str 

286 :return: None 

287 """ 

288 self.remove_description() 

289 self._create_literal(ProvEntity.iri_description, string) 

290 

291 def remove_description(self) -> None: 

292 """ 

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

294 

295 :return: None 

296 """ 

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

298 

299 # IS ATTRIBUTED TO 

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

301 """ 

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

303 

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

305 """ 

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

307 return uri 

308 

309 @accepts_only('thing') 

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

311 """ 

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

313 

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

315 

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

317 

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

319 :type se_agent: URIRef 

320 :return: None 

321 """ 

322 self.remove_resp_agent() 

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

324 

325 def remove_resp_agent(self) -> None: 

326 """ 

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

328 

329 :return: None 

330 """ 

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