Coverage for oc_ocdm / metadata / metadata_entity.py: 81%

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, List 

11 

12from triplelite import RDFTerm, SubgraphView, Triple, TripleLite, rdflib_to_rdfterm 

13 

14from oc_ocdm.abstract_entity import AbstractEntity 

15from oc_ocdm.constants import Namespace 

16 

17if TYPE_CHECKING: 

18 from typing import ClassVar, Dict 

19 

20 from oc_ocdm.metadata.metadata_set import MetadataSet 

21 

22 

23class MetadataEntity(AbstractEntity): 

24 DCTERMS = Namespace("http://purl.org/dc/terms/") 

25 DCAT = Namespace("http://www.w3.org/ns/dcat#") 

26 VOID = Namespace("http://rdfs.org/ns/void#") 

27 

28 iri_dataset = DCAT.Dataset 

29 iri_datafile = DCAT.Distribution 

30 

31 iri_title = DCTERMS.title 

32 iri_description = DCTERMS.description 

33 iri_issued = DCTERMS.issued 

34 iri_modified = DCTERMS.modified 

35 iri_keyword = DCAT.keyword 

36 iri_subject = DCAT.theme 

37 iri_landing_page = DCAT.landingPage 

38 iri_subset = VOID.subset 

39 iri_sparql_endpoint = VOID.sparqlEndpoint 

40 iri_distribution = DCAT.distribution 

41 iri_license = DCTERMS.license 

42 iri_download_url = DCAT.downloadURL 

43 iri_media_type = DCAT.mediaType 

44 iri_byte_size = DCAT.byteSize 

45 

46 short_name_to_type_iri: ClassVar[Dict[str, str]] = {"_dataset_": iri_dataset, "di": iri_datafile} 

47 

48 def __init__( 

49 self, 

50 g: TripleLite, 

51 base_iri: str, 

52 dataset_name: str, 

53 m_set: MetadataSet, 

54 res_type: str, 

55 res: str | None = None, 

56 resp_agent: str | None = None, 

57 source: str | None = None, 

58 count: str | None = None, 

59 label: str | None = None, 

60 short_name: str = "", 

61 preexisting_graph: SubgraphView | None = None, 

62 ) -> None: 

63 super(MetadataEntity, self).__init__() 

64 self.g: TripleLite = g 

65 self.base_iri: str = base_iri 

66 self.dataset_name: str = dataset_name 

67 self.resp_agent: str | None = resp_agent 

68 self.source: str | None = source 

69 self.short_name: str = short_name 

70 self.m_set: MetadataSet = m_set 

71 self._preexisting_triples: frozenset[Triple] = frozenset() 

72 self._merge_list: tuple[MetadataEntity, ...] = () 

73 # FLAGS 

74 self._to_be_deleted: bool = False 

75 self._was_merged: bool = False 

76 

77 # If res was not specified, create from scratch the URI reference for this entity, 

78 # otherwise use the provided one 

79 if res is None: 

80 base_res: str = self.base_iri + self.dataset_name 

81 if base_res[-1] != "/": 

82 base_res += "/" 

83 self.res = self._generate_new_res(count, base_res, short_name) 

84 else: 

85 self.res = res 

86 

87 # If not already done, register this MetadataEntity instance inside the MetadataSet 

88 if self.res not in m_set.res_to_entity: 

89 m_set.res_to_entity[self.res] = self 

90 

91 if preexisting_graph is not None: 

92 self.remove_every_triple() 

93 self._preexisting_triples = frozenset( 

94 (self.res, p, rdflib_to_rdfterm(o)) for p, o in preexisting_graph.predicate_objects(self.res) 

95 ) 

96 self.g.add_many(self._preexisting_triples) 

97 else: 

98 # Add mandatory information to the entity graph 

99 self._create_type(res_type) 

100 if label is not None: 

101 self.create_label(label) 

102 

103 @staticmethod 

104 def _generate_new_res(count: str | None, base_res: str, short_name: str) -> str: 

105 if short_name == "_dataset_": 

106 return base_res 

107 else: 

108 assert count is not None 

109 return base_res + short_name + "/" + count 

110 

111 @property 

112 def to_be_deleted(self) -> bool: 

113 return self._to_be_deleted 

114 

115 @property 

116 def was_merged(self) -> bool: 

117 return self._was_merged 

118 

119 @property 

120 def merge_list(self) -> tuple[MetadataEntity, ...]: 

121 return self._merge_list 

122 

123 @property 

124 def preexisting_triples(self) -> frozenset[Triple]: 

125 return self._preexisting_triples 

126 

127 def mark_as_to_be_deleted(self) -> None: 

128 # Here we must REMOVE triples pointing 

129 # to 'self' [THIS CANNOT BE UNDONE]: 

130 for res, entity in self.m_set.res_to_entity.items(): 

131 triples_list: List[Triple] = list(entity.g.triples((res, None, RDFTerm("uri", str(self.res))))) 

132 for triple in triples_list: 

133 entity.g.remove(triple) 

134 

135 self._to_be_deleted = True 

136 

137 def merge(self, other: object) -> None: 

138 """ 

139 **WARNING:** ``MetadataEntity`` **is an abstract class that cannot be instantiated at runtime. 

140 As such, it's only possible to execute this method on entities generated from** 

141 ``MetadataEntity``'s **subclasses. Please, refer to their documentation of the** `merge` **method.** 

142 

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

144 be merged into the current entity. 

145 :type other: MetadataEntity 

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

147 :return: None 

148 """ 

149 if not isinstance(other, MetadataEntity) or other.short_name != self.short_name: 

150 raise TypeError( 

151 f"[{self.__class__.__name__}.merge] Expected entity type: {self.short_name}. " 

152 f"Provided: {type(other).__name__}." 

153 ) 

154 

155 # Here we must REDIRECT triples pointing 

156 # to 'other' to make them point to 'self': 

157 for res, entity in self.m_set.res_to_entity.items(): 

158 triples_list: List[Triple] = list(entity.g.triples((res, None, RDFTerm("uri", str(other.res))))) 

159 for triple in triples_list: 

160 entity.g.remove(triple) 

161 new_triple = (triple[0], triple[1], RDFTerm("uri", str(self.res))) 

162 entity.g.add(new_triple) 

163 

164 types: List[str] = other.get_types() 

165 for cur_type in types: 

166 self._create_type(cur_type) 

167 

168 label: str | None = other.get_label() 

169 if label is not None: 

170 self.create_label(label) 

171 

172 self._was_merged = True 

173 self._merge_list = (*self._merge_list, other) 

174 

175 # 'other' must be deleted AFTER the redirection of 

176 # triples pointing to it, since mark_as_to_be_deleted 

177 # also removes every triple pointing to 'other' 

178 other.mark_as_to_be_deleted() 

179 

180 self._merge_properties(other) 

181 

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

183 pass 

184 

185 def commit_changes(self) -> None: 

186 if self._to_be_deleted: 

187 self._preexisting_triples = frozenset() 

188 self.remove_every_triple() 

189 else: 

190 self._preexisting_triples = frozenset(self.g.triples((self.res, None, None))) 

191 self._to_be_deleted = False 

192 self._was_merged = False 

193 self._merge_list = ()