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

78 statements  

« prev     ^ index     » next       coverage.py v7.13.4, created at 2026-05-08 20:23 +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 triplelite import RDFTerm 

11 

12from oc_ocdm.constants import XSD_DATETIME, XSD_DECIMAL 

13from oc_ocdm.metadata.metadata_entity import MetadataEntity 

14 

15 

16class Distribution(MetadataEntity): 

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

18 file.""" 

19 

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

21 """ 

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

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

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

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

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

27 

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

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

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

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

32 

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

34 be merged into the current entity. 

35 :type other: Distribution 

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

37 :return: None 

38 """ 

39 super()._merge_properties(other) 

40 assert isinstance(other, Distribution) 

41 

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

43 if title is not None: 

44 self.has_title(title) 

45 

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

47 if description is not None: 

48 self.has_description(description) 

49 

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

51 if pub_date is not None: 

52 self.has_publication_date(pub_date) 

53 

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

55 if byte_size is not None: 

56 self.has_byte_size(byte_size) 

57 

58 license_uri: str | None = other.get_license() 

59 if license_uri is not None: 

60 self.has_license(license_uri) 

61 

62 download_url: str | None = other.get_download_url() 

63 if download_url is not None: 

64 self.has_download_url(download_url) 

65 

66 media_type: str | None = other.get_media_type() 

67 if media_type is not None: 

68 self.has_media_type(media_type) 

69 

70 # HAS TITLE 

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

72 """ 

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

74 

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

76 """ 

77 return self._get_literal(MetadataEntity.iri_title) 

78 

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

80 """ 

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

82 

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

84 

85 `The title of the distribution.` 

86 

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

88 :type string: str 

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

90 :return: None 

91 """ 

92 self.remove_title() 

93 self._create_literal(MetadataEntity.iri_title, string) 

94 

95 def remove_title(self) -> None: 

96 """ 

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

98 

99 :return: None 

100 """ 

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

102 

103 # HAS DESCRIPTION 

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

105 """ 

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

107 

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

109 """ 

110 return self._get_literal(MetadataEntity.iri_description) 

111 

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

113 """ 

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

115 

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

117 

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

119 

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

121 :type string: str 

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

123 :return: None 

124 """ 

125 self.remove_description() 

126 self._create_literal(MetadataEntity.iri_description, string) 

127 

128 def remove_description(self) -> None: 

129 """ 

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

131 

132 :return: None 

133 """ 

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

135 

136 # HAS PUBLICATION DATE 

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

138 """ 

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

140 

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

142 """ 

143 return self._get_literal(MetadataEntity.iri_issued) 

144 

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

146 """ 

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

148 

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

150 

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

152 

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

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

155 :type string: str 

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

157 :return: None 

158 """ 

159 self.remove_publication_date() 

160 self._create_literal(MetadataEntity.iri_issued, string, XSD_DATETIME, False) 

161 

162 def remove_publication_date(self) -> None: 

163 """ 

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

165 

166 :return: None 

167 """ 

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

169 

170 # HAS BYTE SIZE 

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

172 """ 

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

174 

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

176 """ 

177 return self._get_literal(MetadataEntity.iri_byte_size) 

178 

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

180 """ 

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

182 

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

184 

185 `The size in bytes of the distribution.` 

186 

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

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

189 :type string: str 

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

191 :return: None 

192 """ 

193 self.remove_byte_size() 

194 self._create_literal(MetadataEntity.iri_byte_size, string, XSD_DECIMAL) 

195 

196 def remove_byte_size(self) -> None: 

197 """ 

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

199 

200 :return: None 

201 """ 

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

203 

204 # HAS LICENSE 

205 def get_license(self) -> str | None: 

206 """ 

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

208 

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

210 """ 

211 return self._get_uri_reference(MetadataEntity.iri_license) 

212 

213 def has_license(self, thing_res: str) -> None: 

214 """ 

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

216 

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

218 

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

220 

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

222 :type thing_res: URIRef 

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

224 :return: None 

225 """ 

226 self.remove_license() 

227 self.g.add((self.res, MetadataEntity.iri_license, RDFTerm("uri", str(thing_res)))) 

228 

229 def remove_license(self) -> None: 

230 """ 

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

232 

233 :return: None 

234 """ 

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

236 

237 # HAS DOWNLOAD URL 

238 def get_download_url(self) -> str | None: 

239 """ 

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

241 

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

243 """ 

244 return self._get_uri_reference(MetadataEntity.iri_download_url) 

245 

246 def has_download_url(self, thing_res: str) -> None: 

247 """ 

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

249 

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

251 

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

253 

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

255 :type thing_res: URIRef 

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

257 :return: None 

258 """ 

259 self.remove_download_url() 

260 self.g.add((self.res, MetadataEntity.iri_download_url, RDFTerm("uri", str(thing_res)))) 

261 

262 def remove_download_url(self) -> None: 

263 """ 

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

265 

266 :return: None 

267 """ 

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

269 

270 # HAS_MEDIA_TYPE 

271 def get_media_type(self) -> str | None: 

272 """ 

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

274 

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

276 """ 

277 return self._get_uri_reference(MetadataEntity.iri_media_type) 

278 

279 def has_media_type(self, thing_res: str) -> None: 

280 """ 

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

282 

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

284 

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

286 

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

288 :type thing_res: URIRef 

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

290 :return: None 

291 """ 

292 self.remove_media_type() 

293 self.g.add((self.res, MetadataEntity.iri_media_type, RDFTerm("uri", str(thing_res)))) 

294 

295 def remove_media_type(self) -> None: 

296 """ 

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

298 

299 :return: None 

300 """ 

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