Coverage for oc_ocdm / metadata / metadata_set.py: 73%

1#!/usr/bin/python 

2 

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

4# 

5# SPDX-License-Identifier: ISC 

6 

7# -*- coding: utf-8 -*- 

8from __future__ import annotations 

9 

10from typing import TYPE_CHECKING, cast 

11 

12from oc_ocdm.counter_handler.counter_handler import CounterHandler 

13from oc_ocdm.counter_handler.filesystem_counter_handler import FilesystemCounterHandler 

14from oc_ocdm.counter_handler.in_memory_counter_handler import InMemoryCounterHandler 

15from oc_ocdm.metadata.entities.dataset import Dataset 

16from oc_ocdm.metadata.entities.distribution import Distribution 

17from oc_ocdm.support.support import get_count, get_short_name, is_dataset 

18 

19if TYPE_CHECKING: 

20 from typing import ClassVar, Dict, Tuple 

21 

22from triplelite import SubgraphView, TripleLite 

23 

24from oc_ocdm.abstract_set import AbstractSet 

25from oc_ocdm.metadata.metadata_entity import MetadataEntity 

26 

27 

28class MetadataSet(AbstractSet[MetadataEntity]): 

29 # Labels 

30 labels: ClassVar[Dict[str, str]] = {"_dataset_": "dataset", "di": "distribution"} 

31 

32 def __init__(self, base_iri: str, info_dir: str | None = "", wanted_label: bool = True) -> None: 

33 super(MetadataSet, self).__init__() 

34 # The following variable maps a str with the related metadata entity 

35 self.res_to_entity: Dict[str, MetadataEntity] = {} 

36 self.base_iri: str = base_iri 

37 if self.base_iri[-1] != "/": 

38 self.base_iri += "/" 

39 self.wanted_label: bool = wanted_label 

40 

41 if info_dir is not None and info_dir != "": 

42 self.counter_handler: CounterHandler = FilesystemCounterHandler(info_dir) 

43 else: 

44 self.counter_handler: CounterHandler = InMemoryCounterHandler() 

45 

46 def get_entity(self, res: str) -> MetadataEntity | None: 

47 if res in self.res_to_entity: 

48 return self.res_to_entity[res] 

49 

50 def add_dataset( 

51 self, 

52 dataset_name: str, 

53 resp_agent: str, 

54 source: str | None = None, 

55 res: str | None = None, 

56 preexisting_graph: SubgraphView | None = None, 

57 ) -> Dataset: 

58 if res is not None and not is_dataset(res): 

59 raise ValueError(f"Given res: <{res}> is inappropriate for a Dataset entity.") 

60 if res is not None and res in self.res_to_entity: 

61 return cast(Dataset, self.res_to_entity[res]) 

62 # Here we use a fictitious short name for Dataset, since the OCDM document doesn't specify 

63 # any particular short name for this type of entity. It's only used internally to distinguish 

64 # between different metadata entities but it's meaningless outside of this scope. 

65 cur_g, count, label = self._add_metadata("_dataset_", dataset_name, res) 

66 return Dataset( 

67 cur_g, 

68 self.base_iri, 

69 dataset_name, 

70 self, 

71 MetadataEntity.iri_dataset, 

72 res, 

73 resp_agent, 

74 source, 

75 count, 

76 label, 

77 "_dataset_", 

78 preexisting_graph, 

79 ) 

80 

81 def add_di( 

82 self, 

83 dataset_name: str, 

84 resp_agent: str, 

85 source: str | None = None, 

86 res: str | None = None, 

87 preexisting_graph: SubgraphView | None = None, 

88 ) -> Distribution: 

89 if res is not None and get_short_name(res) != "di": 

90 raise ValueError(f"Given res: <{res}> is inappropriate for a Distribution entity.") 

91 if res is not None and res in self.res_to_entity: 

92 return cast(Distribution, self.res_to_entity[res]) 

93 cur_g, count, label = self._add_metadata("di", dataset_name, res) 

94 return Distribution( 

95 cur_g, 

96 self.base_iri, 

97 dataset_name, 

98 self, 

99 MetadataEntity.iri_datafile, 

100 res, 

101 resp_agent, 

102 source, 

103 count, 

104 label, 

105 "di", 

106 preexisting_graph, 

107 ) 

108 

109 def _add_metadata( 

110 self, short_name: str, dataset_name: str, res: str | None = None 

111 ) -> Tuple[TripleLite, str | None, str | None]: 

112 cur_g = TripleLite() 

113 

114 count: str | None = None 

115 label: str | None = None 

116 

117 if res is not None: 

118 if short_name != "_dataset_": # Datasets don't have a counter associated with them... 

119 try: 

120 res_count: int = int(get_count(res)) 

121 except ValueError: 

122 res_count: int = -1 

123 if res_count > self.counter_handler.read_metadata_counter(short_name, dataset_name): 

124 self.counter_handler.set_metadata_counter(res_count, short_name, dataset_name) 

125 return cur_g, count, label 

126 

127 if short_name != "_dataset_": # Datasets don't have a counter associated with them... 

128 count = str(self.counter_handler.increment_metadata_counter(short_name, dataset_name)) 

129 

130 if self.wanted_label: 

131 label = "%s %s [%s/%s]" % (self.labels[short_name], count, short_name, count) 

132 

133 return cur_g, count, label 

134 

135 def commit_changes(self): 

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

137 entity.commit_changes() 

138 if entity.to_be_deleted: 

139 del self.res_to_entity[res] 

140 

141 def get_dataset(self) -> tuple[Dataset, ...]: 

142 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, Dataset)) 

143 

144 def get_di(self) -> tuple[Distribution, ...]: 

145 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, Distribution))