Coverage for oc_ocdm / metadata / entities / distribution.py: 97%

86 statements  

« prev     ^ index     » next       coverage.py v7.13.4, created at 2026-03-28 18:52 +0000

1#!/usr/bin/python 

2 

3# SPDX-FileCopyrightText: 2020-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 rdflib import XSD 

13 

14from oc_ocdm.decorators import accepts_only 

15from oc_ocdm.metadata.metadata_entity import MetadataEntity 

16 

17if TYPE_CHECKING: 

18 from rdflib import URIRef 

19 

20 

21class Distribution(MetadataEntity): 

22 """Distribution (short: di): an accessible form of a dataset, for example a downloadable 

23 file.""" 

24 

25 def _merge_properties(self, other: MetadataEntity) -> None: 

26 """ 

27 The merge operation allows combining two ``Distribution`` entities into a single one, 

28 by marking the second entity as to be deleted while also copying its data into the current 

29 ``Distribution``. Moreover, every triple from the containing ``MetadataSet`` referring to the second 

30 entity gets "redirected" to the current entity: **every other reference contained inside a 

31 different source (e.g. a triplestore) must be manually handled by the user!** 

32 

33 In case of functional properties, values from the current entity get overwritten 

34 by those coming from the second entity while, in all other cases, values from the 

35 second entity are simply appended to those of the current entity. In this context, 

36 ``rdfs:label`` is considered as a functional property, while ``rdf:type`` is not. 

37 

38 :param other: The entity which will be marked as to be deleted and whose properties will 

39 be merged into the current entity. 

40 :type other: Distribution 

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

42 :return: None 

43 """ 

44 super()._merge_properties(other) 

45 assert isinstance(other, Distribution) 

46 

47 title: str | None = other.get_title() 

48 if title is not None: 

49 self.has_title(title) 

50 

51 description: str | None = other.get_description() 

52 if description is not None: 

53 self.has_description(description) 

54 

55 pub_date: str | None = other.get_publication_date() 

56 if pub_date is not None: 

57 self.has_publication_date(pub_date) 

58 

59 byte_size: str | None = other.get_byte_size() 

60 if byte_size is not None: 

61 self.has_byte_size(byte_size) 

62 

63 license_uri: URIRef | None = other.get_license() 

64 if license_uri is not None: 

65 self.has_license(license_uri) 

66 

67 download_url: URIRef | None = other.get_download_url() 

68 if download_url is not None: 

69 self.has_download_url(download_url) 

70 

71 media_type: URIRef | None = other.get_media_type() 

72 if media_type is not None: 

73 self.has_media_type(media_type) 

74 

75 # HAS TITLE 

76 def get_title(self) -> str | None: 

77 """ 

78 Getter method corresponding to the ``dcterms:title`` RDF predicate. 

79 

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

81 """ 

82 return self._get_literal(MetadataEntity.iri_title) 

83 

84 @accepts_only('literal') 

85 def has_title(self, string: str) -> None: 

86 """ 

87 Setter method corresponding to the ``dcterms:title`` RDF predicate. 

88 

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

90 

91 `The title of the distribution.` 

92 

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

94 :type string: str 

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

96 :return: None 

97 """ 

98 self.remove_title() 

99 self._create_literal(MetadataEntity.iri_title, string) 

100 

101 def remove_title(self) -> None: 

102 """ 

103 Remover method corresponding to the ``dcterms:title`` RDF predicate. 

104 

105 :return: None 

106 """ 

107 self.g.remove((self.res, MetadataEntity.iri_title, None)) 

108 

109 # HAS DESCRIPTION 

110 def get_description(self) -> str | None: 

111 """ 

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

113 

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

115 """ 

116 return self._get_literal(MetadataEntity.iri_description) 

117 

118 @accepts_only('literal') 

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

120 """ 

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

122 

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

124 

125 `A short textual description of the content of the distribution.` 

126 

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

128 :type string: str 

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

130 :return: None 

131 """ 

132 self.remove_description() 

133 self._create_literal(MetadataEntity.iri_description, string) 

134 

135 def remove_description(self) -> None: 

136 """ 

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

138 

139 :return: None 

140 """ 

141 self.g.remove((self.res, MetadataEntity.iri_description, None)) 

142 

143 # HAS PUBLICATION DATE 

144 def get_publication_date(self) -> str | None: 

145 """ 

146 Getter method corresponding to the ``dcterms:issued`` RDF predicate. 

147 

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

149 """ 

150 return self._get_literal(MetadataEntity.iri_issued) 

151 

152 @accepts_only('literal') 

153 def has_publication_date(self, string: str) -> None: 

154 """ 

155 Setter method corresponding to the ``dcterms:issued`` RDF predicate. 

156 

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

158 

159 `The date of first publication of the distribution.` 

160 

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

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

163 :type string: str 

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

165 :return: None 

166 """ 

167 self.remove_publication_date() 

168 self._create_literal(MetadataEntity.iri_issued, string, XSD.dateTime, False) 

169 

170 def remove_publication_date(self) -> None: 

171 """ 

172 Remover method corresponding to the ``dcterms:issued`` RDF predicate. 

173 

174 :return: None 

175 """ 

176 self.g.remove((self.res, MetadataEntity.iri_issued, None)) 

177 

178 # HAS BYTE SIZE 

179 def get_byte_size(self) -> str | None: 

180 """ 

181 Getter method corresponding to the ``dcat:byte_size`` RDF predicate. 

182 

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

184 """ 

185 return self._get_literal(MetadataEntity.iri_byte_size) 

186 

187 @accepts_only('literal') 

188 def has_byte_size(self, string: str) -> None: 

189 """ 

190 Setter method corresponding to the ``dcat:byte_size`` RDF predicate. 

191 

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

193 

194 `The size in bytes of the distribution.` 

195 

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

197 be a string compliant with the** ``xsd:decimal`` **datatype.** 

198 :type string: str 

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

200 :return: None 

201 """ 

202 self.remove_byte_size() 

203 self._create_literal(MetadataEntity.iri_byte_size, string, XSD.decimal) 

204 

205 def remove_byte_size(self) -> None: 

206 """ 

207 Remover method corresponding to the ``dcat:byte_size`` RDF predicate. 

208 

209 :return: None 

210 """ 

211 self.g.remove((self.res, MetadataEntity.iri_byte_size, None)) 

212 

213 # HAS LICENSE 

214 def get_license(self) -> URIRef | None: 

215 """ 

216 Getter method corresponding to the ``dcterms:license`` RDF predicate. 

217 

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

219 """ 

220 return self._get_uri_reference(MetadataEntity.iri_license) 

221 

222 @accepts_only('thing') 

223 def has_license(self, thing_res: URIRef) -> None: 

224 """ 

225 Setter method corresponding to the ``dcterms:license`` RDF predicate. 

226 

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

228 

229 `The resource describing the license associated with the data in the distribution.` 

230 

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

232 :type thing_res: URIRef 

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

234 :return: None 

235 """ 

236 self.remove_license() 

237 self.g.add((self.res, MetadataEntity.iri_license, thing_res)) 

238 

239 def remove_license(self) -> None: 

240 """ 

241 Remover method corresponding to the ``dcterms:license`` RDF predicate. 

242 

243 :return: None 

244 """ 

245 self.g.remove((self.res, MetadataEntity.iri_license, None)) 

246 

247 # HAS DOWNLOAD URL 

248 def get_download_url(self) -> URIRef | None: 

249 """ 

250 Getter method corresponding to the ``dcat:downloadURL`` RDF predicate. 

251 

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

253 """ 

254 return self._get_uri_reference(MetadataEntity.iri_download_url) 

255 

256 @accepts_only('thing') 

257 def has_download_url(self, thing_res: URIRef) -> None: 

258 """ 

259 Setter method corresponding to the ``dcat:downloadURL`` RDF predicate. 

260 

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

262 

263 `The URL of the document where the distribution is stored.` 

264 

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

266 :type thing_res: URIRef 

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

268 :return: None 

269 """ 

270 self.remove_download_url() 

271 self.g.add((self.res, MetadataEntity.iri_download_url, thing_res)) 

272 

273 def remove_download_url(self) -> None: 

274 """ 

275 Remover method corresponding to the ``dcat:downloadURL`` RDF predicate. 

276 

277 :return: None 

278 """ 

279 self.g.remove((self.res, MetadataEntity.iri_download_url, None)) 

280 

281 # HAS_MEDIA_TYPE 

282 def get_media_type(self) -> URIRef | None: 

283 """ 

284 Getter method corresponding to the ``dcat:mediaType`` RDF predicate. 

285 

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

287 """ 

288 return self._get_uri_reference(MetadataEntity.iri_media_type) 

289 

290 @accepts_only('thing') 

291 def has_media_type(self, thing_res: URIRef) -> None: 

292 """ 

293 Setter method corresponding to the ``dcat:mediaType`` RDF predicate. 

294 

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

296 

297 `The file type of the representation of the distribution (according to IANA media types).` 

298 

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

300 :type thing_res: URIRef 

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

302 :return: None 

303 """ 

304 self.remove_media_type() 

305 self.g.add((self.res, MetadataEntity.iri_media_type, thing_res)) 

306 

307 def remove_media_type(self) -> None: 

308 """ 

309 Remover method corresponding to the ``dcat:mediaType`` RDF predicate. 

310 

311 :return: None 

312 """ 

313 self.g.remove((self.res, MetadataEntity.iri_media_type, None))