Coverage for rdflib_ocdm/prov/snapshot

1#!/usr/bin/python

3# SPDX-FileCopyrightText: 2016 Silvio Peroni <essepuntato@gmail.com>

4# SPDX-FileCopyrightText: 2023-2026 Arcangelo Massari <arcangelo.massari@unibo.it>

6# SPDX-License-Identifier: ISC

7from __future__ import annotations

9from typing import TYPE_CHECKING

11from rdflib import XSD, URIRef

13from rdflib_ocdm.prov.prov_entity import ProvEntity

15if TYPE_CHECKING:

16 from typing import List, Optional

19class SnapshotEntity(ProvEntity):

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

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

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

23 created or modified the entity metadata."""

25 # HAS CREATION DATE

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

27 """

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

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

31 """

32 return self._get_literal(ProvEntity.iri_generated_at_time)

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

35 """

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

38 **WARNING: this is a functional property, hence

39 any existing value will be overwritten!**

41 `The date on which a particular snapshot of a

42 bibliographic entity's metadata was created.`

44 :param string: The value that will be set as the

45 object of the property related to this method.

46 **It must be a string compliant with the**

47 ``xsd:dateTime`` **datatype.**

48 :type string: str

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

50 :return: None

51 """

52 self.remove_generation_time()

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

55 def remove_generation_time(self) -> None:

56 """

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

59 :return: None

60 """

61 self.g.remove((self.res, ProvEntity.iri_generated_at_time, None)) # type: ignore[arg-type]

63 # HAS INVALIDATION DATE

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

65 """

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

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

69 """

70 return self._get_literal(ProvEntity.iri_invalidated_at_time)

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

73 """

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

76 **WARNING: this is a functional property, hence

77 any existing value will be overwritten!**

79 `The date on which a snapshot of a bibliographic

80 entity's metadata was invalidated due to an

81 update (e.g. a correction, or the addition of

82 some metadata that was not specified in the

83 previous snapshot), or due to a merger of the

84 entity with another one.`

86 :param string: The value that will be set as the

87 object of the property related to this method.

88 **It must be a string compliant with the**

89 ``xsd:dateTime`` **datatype.**

90 :type string: str

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

92 :return: None

93 """

94 self.remove_invalidation_time()

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

97 def remove_invalidation_time(self) -> None:

98 """

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

100

101 :return: None

102 """

103 self.g.remove((self.res, ProvEntity.iri_invalidated_at_time, None)) # type: ignore[arg-type]

104

105 # IS SNAPSHOT OF

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

107 """

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

109

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

111 """

112 uri: Optional[URIRef] = self._get_uri_reference(

113 ProvEntity.iri_specialization_of

114 )

115 return uri

116

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

118 """

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

120

121 **WARNING: this is a functional property, hence

122 any existing value will be overwritten!**

123

124 `This property is used to link a snapshot of

125 entity metadata to the bibliographic entity

126 to which the snapshot refers.`

127

128 :param en_res: The value that will be set as

129 the object of the property related to this

130 method

131 :type en_res: URIRef

132 :return: None

133 """

134 self.remove_is_snapshot_of()

135 self.g.add((self.res, ProvEntity.iri_specialization_of, en_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)) # type: ignore[arg-type]

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(

153 ProvEntity.iri_was_derived_from

154 )

155 result: List[ProvEntity] = []

156 for uri in uri_list:

157 prov_subj = uri.split("/prov/se/")[0]

158 result.append(self.g.add_se(URIRef(prov_subj), uri)) # type: ignore[union-attr]

159 return result

160

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

162 """

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

164

165 `This property is used to identify the

166 immediately previous snapshot of entity metadata

167 associated with the same bibliographic entity.`

168

169 :param se_res: The value that will be set as

170 the object of the property related to this

171 method

172 :type se_res: ProvEntity

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

174 :return: None

175 """

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

177

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

179 """

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

181

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

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

184

185 :param se_res: If not None, the specific object

186 value that will be removed from the property

187 related to this method (defaults to None)

188 :type se_res: SnapshotEntity

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

190 :return: None

191 """

192 if se_res is not None:

193 self.g.remove((self.res, ProvEntity.iri_was_derived_from, se_res.res)) # type: ignore[arg-type]

194 else:

195 self.g.remove((self.res, ProvEntity.iri_was_derived_from, None)) # type: ignore[arg-type]

196

197 # HAS PRIMARY SOURCE

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

199 """

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

201

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

203 """

204 uri: Optional[URIRef] = self._get_uri_reference(

205 ProvEntity.iri_had_primary_source

206 )

207 return uri

208

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

210 """

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

212

213 **WARNING: this is a functional property, hence

214 any existing value will be overwritten!**

215

216 `This property is used to identify the primary

217 source from which the metadata described in the

218 snapshot are derived (e.g. Crossref, as the

219 result of querying the CrossRef API).`

220

221 :param any_res: The value that will be set as

222 the object of the property related to this

223 method

224 :type any_res: URIRef

225 :return: None

226 """

227 self.remove_primary_source()

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

229

230 def remove_primary_source(self) -> None:

231 """

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

233

234 :return: None

235 """

236 self.g.remove((self.res, ProvEntity.iri_had_primary_source, None)) # type: ignore[arg-type]

237

238 # HAS UPDATE ACTION

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

240 """

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

242

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

244 """

245 return self._get_literal(ProvEntity.iri_has_update_query)

246

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

248 """

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

250

251 **WARNING: this is a functional property, hence

252 any existing value will be overwritten!**

253

254 `The UPDATE SPARQL query that specifies which

255 data, associated to the bibliographic entity in

256 consideration, have been modified (e.g. for

257 correcting a mistake) in the current snapshot

258 starting from those associated to the previous

259 snapshot of the entity.`

260

261 :param string: The value that will be set as

262 the object of the property related to this

263 method

264 :type string: str

265 :return: None

266 """

267 self.remove_update_action()

268 self._create_literal(ProvEntity.iri_has_update_query, string)

269

270 def remove_update_action(self) -> None:

271 """

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

273

274 :return: None

275 """

276 self.g.remove((self.res, ProvEntity.iri_has_update_query, None)) # type: ignore[arg-type]

277

278 # HAS DESCRIPTION

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

280 """

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

282

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

284 """

285 return self._get_literal(ProvEntity.iri_description)

286

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

288 """

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

290

291 **WARNING: this is a functional property, hence

292 any existing value will be overwritten!**

293

294 `A textual description of the events that have

295 resulted in the current snapshot (e.g. the

296 creation of the initial snapshot, the creation

297 of a new snapshot following the modification of

298 the entity to which the metadata relate, or the

299 creation of a new snapshot following the merger

300 with another entity of the entity to which the

301 previous snapshot related).`

302

303 :param string: The value that will be set as

304 the object of the property related to this

305 method

306 :type string: str

307 :return: None

308 """

309 self.remove_description()

310 self._create_literal(ProvEntity.iri_description, string)

311

312 def remove_description(self) -> None:

313 """

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

315

316 :return: None

317 """

318 self.g.remove((self.res, ProvEntity.iri_description, None)) # type: ignore[arg-type]

319

320 # IS ATTRIBUTED TO

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

322 """

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

324

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

326 """

327 uri: Optional[URIRef] = self._get_uri_reference(

328 ProvEntity.iri_was_attributed_to

329 )

330 return uri

331

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

333 """

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

335

336 **WARNING: this is a functional property, hence

337 any existing value will be overwritten!**

338

339 `The agent responsible for the creation of the

340 current entity snapshot.`

341

342 :param se_agent: The value that will be set as

343 the object of the property related to this

344 method

345 :type se_agent: URIRef

346 :return: None

347 """

348 self.remove_resp_agent()

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

350

351 def remove_resp_agent(self) -> None:

352 """

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

354

355 :return: None

356 """

357 self.g.remove((self.res, ProvEntity.iri_was_attributed_to, None)) # type: ignore[arg-type]

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

70 statements