Coverage for oc_ds_converter / oc_idmanager / oc_data_storage / in_memory_manager.py: 80%

1# SPDX-FileCopyrightText: 2023 Arianna Moretti <arianna.moretti4@unibo.it> 

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

3# 

4# SPDX-License-Identifier: ISC 

5 

6from __future__ import annotations 

7 

8import json 

9import os 

10from pathlib import Path 

11 

12from oc_ds_converter.oc_idmanager.oc_data_storage.storage_manager import StorageManager 

13 

14 

15class InMemoryStorageManager(StorageManager): 

16 """A concrete implementation of the ``StorageManager`` interface that persistently stores 

17 the IDs validity values within a in-memory dictionary, which is eventually saved in a json file.""" 

18 

19 def __init__(self, json_file_path: str | None = None, **params: object) -> None: 

20 """ 

21 Constructor of the ``InMemoryStorageManager`` class. 

22 """ 

23 super().__init__(**params) 

24 if json_file_path and os.path.exists(json_file_path): 

25 self.storage_filepath = json_file_path 

26 o_jfp = open(self.storage_filepath, "r") 

27 self.id_value_dict = json.load(o_jfp) 

28 o_jfp.close() 

29 elif json_file_path and not os.path.exists(json_file_path): 

30 if not os.path.exists(os.path.abspath(os.path.join(json_file_path, os.pardir))): 

31 Path(os.path.abspath(os.path.join(json_file_path, os.pardir))).mkdir(parents=True, exist_ok=True) 

32 self.storage_filepath = json_file_path 

33 self.id_value_dict = dict() 

34 file = open(self.storage_filepath, "w", encoding='utf8') 

35 json.dump(self.id_value_dict, file) 

36 file.close() 

37 else: 

38 new_path_dir = os.path.join(os.getcwd(), "storage") 

39 if not os.path.exists(new_path_dir): 

40 os.makedirs(new_path_dir) 

41 filepath = os.path.join(new_path_dir, "id_value.json" ) 

42 self.id_value_dict = dict() 

43 self.storage_filepath = filepath 

44 file = open(self.storage_filepath, "w", encoding='utf8') 

45 json.dump(self.id_value_dict, file) 

46 file.close() 

47 

48 def set_full_value(self, id: str, value: dict[str, str | bool | object]) -> None: 

49 """ 

50 It allows to set the counter value of provenance entities. 

51 

52 :param value: The new counter value to be set 

53 :type value: dict 

54 :param id: The id string with prefix 

55 :type id: str 

56 :raises ValueError: if ``value`` is neither 0 nor 1 (0 is False, 1 is True). 

57 :return: None 

58 """ 

59 id_name = str(id) 

60 if not isinstance(value, dict): 

61 raise ValueError("value must be dict") 

62 if id_name in self.id_value_dict: 

63 new_info = {k: v for k, v in value.items() if k not in self.id_value_dict[id_name]} 

64 self.id_value_dict[id_name].update(new_info) 

65 else: 

66 self.id_value_dict[id_name] = dict(value) 

67 

68 def set_value(self, id: str, value: bool) -> None: 

69 """ 

70 It allows to set the counter value of provenance entities. 

71 

72 :param value: The new counter value to be set 

73 :type value: bool 

74 :param id: The id string with prefix 

75 :type id: str 

76 :raises ValueError: if ``value`` is neither 0 nor 1 (0 is False, 1 is True). 

77 :return: None 

78 """ 

79 id_name = str(id) 

80 if not isinstance(value, bool): 

81 raise ValueError("value must be boolean") 

82 if id_name in self.id_value_dict: 

83 self.id_value_dict[id_name]["valid"] = value 

84 else: 

85 self.id_value_dict[id_name] = {"valid": value} 

86 

87 def get_value(self, id: str) -> bool | None: 

88 """ 

89 It allows to read the value of the "valid" key of the identifier's dict. 

90 

91 :param id: The id name 

92 :type id: str 

93 :return: The requested id value. 

94 """ 

95 id_name = str(id) 

96 id_in_dict = self.id_value_dict.get(id_name) 

97 if id_in_dict: 

98 return id_in_dict["valid"] 

99 return None 

100 

101 def store_file(self) -> None: 

102 """ 

103 It stores in a file the dictionary with the validation results 

104 """ 

105 with open(self.storage_filepath, "w", encoding='utf8') as file: 

106 json.dump(self.id_value_dict, file, indent=4) 

107 

108 def delete_storage(self) -> None: 

109 self.id_value_dict = {} 

110 if os.path.exists(self.storage_filepath): 

111 os.remove(self.storage_filepath) 

112 

113 def get_all_keys(self) -> list[str]: 

114 return list(self.id_value_dict.keys()) 

115 

116 def get_validity_dict(self) -> dict[str, bool]: 

117 return {k: v["valid"] for k, v in self.id_value_dict.items()} 

118 

119 def get_validity_list_of_tuples(self) -> list[tuple[str, bool]]: 

120 return [(k, v["valid"]) for k, v in self.id_value_dict.items()]