Coverage for lib/datamodel/datasource.py: 82%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3 

4# Hermes : Change Data Capture (CDC) tool from any source(s) to any target 

5# Copyright (C) 2023, 2024 INSA Strasbourg 

6# 

7# This file is part of Hermes. 

8# 

9# Hermes is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# Hermes is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with Hermes. If not, see <https://www.gnu.org/licenses/>. 

21 

22 

23from collections.abc import KeysView, ValuesView, ItemsView 

24from typing import Any, Iterator 

25 

26from lib.datamodel.dataobjectlist import DataObjectList 

27from lib.datamodel.dataschema import Dataschema 

28 

29from copy import deepcopy 

30 

31 

32class Datasource: 

33 """Generic data source offering basic methods for data access 

34 Also offers optional trashbin and cache management 

35 """ 

36 

37 def __init__( 

38 self, 

39 schema: Dataschema, 

40 enableTrashbin: bool = False, 

41 enableCache: bool = True, 

42 cacheFilePrefix: str = "", 

43 cacheFileSuffix: str = "", 

44 ): 

45 self._hasTrashbin: bool = enableTrashbin 

46 self._hasCache: bool = enableCache 

47 self.__cacheFilePrefix: str = cacheFilePrefix 

48 self.__cacheFileSuffix: str = cacheFileSuffix 

49 self.schema: Dataschema = schema 

50 """Copy of Dataschema used to build current datasource""" 

51 

52 self._data: dict[str, DataObjectList] = {} 

53 """Dictionary containing the datamodel object types specified in server 

54 datamodel or in client schema with object name as key, and their corresponding 

55 DataObjectList as value""" 

56 

57 for objtype, objlistcls in self.schema.objectlistTypes.items(): 

58 self._data[objtype] = objlistcls(from_json_dict=[]) 

59 

60 if self._hasTrashbin: 

61 for objtype, objlistcls in self.schema.objectlistTypes.items(): 

62 self._data["trashbin_" + objtype] = objlistcls(from_json_dict=[]) 

63 

64 if self._hasCache: 

65 self._cache: Datasource = Datasource( 

66 schema=self.schema, 

67 enableTrashbin=self._hasTrashbin, 

68 enableCache=False, 

69 cacheFilePrefix=cacheFilePrefix, 

70 cacheFileSuffix=cacheFileSuffix, 

71 ) 

72 self._cache.loadFromCache() 

73 

74 @property 

75 def cache(self) -> "Datasource": 

76 """Returns the cache""" 

77 if self._hasCache: 

78 return self._cache 

79 raise AttributeError("Asking for cache on an instance with cache disabled") 

80 

81 def loadFromCache(self): 

82 for objtype, objlistcls in self.schema.objectlistTypes.items(): 

83 self._data[objtype] = objlistcls.loadcachefile( 

84 f"{self.__cacheFilePrefix}{objtype}{self.__cacheFileSuffix}" 

85 ) 

86 

87 if self._hasTrashbin: 

88 for objtype, objlistcls in self.schema.objectlistTypes.items(): 

89 self._data["trashbin_" + objtype] = objlistcls.loadcachefile( 

90 f"{self.__cacheFilePrefix}trashbin_" 

91 f"{objtype}{self.__cacheFileSuffix}" 

92 ) 

93 

94 def save(self): 

95 for objtype in self.schema.objectlistTypes: 

96 self._data[objtype].savecachefile( 

97 f"{self.__cacheFilePrefix}{objtype}{self.__cacheFileSuffix}" 

98 ) 

99 

100 if self._hasTrashbin: 

101 for objtype in self.schema.objectlistTypes: 

102 self._data["trashbin_" + objtype].savecachefile( 

103 f"{self.__cacheFilePrefix}trashbin_" 

104 f"{objtype}{self.__cacheFileSuffix}" 

105 ) 

106 

107 def __len__(self) -> int: 

108 return self._data.__len__() 

109 

110 def __getitem__(self, key: str) -> DataObjectList: 

111 return self._data.__getitem__(key) 

112 

113 def __setitem__(self, key: str, value: DataObjectList): 

114 self._data.__setitem__(key, value) 

115 

116 def __delitem__(self, key: str): 

117 self._data.__delitem__(key) 

118 

119 def __iter__(self) -> Iterator[str]: 

120 return self._data.__iter__() 

121 

122 def __reversed__(self) -> Iterator[str]: 

123 return self._data.__reversed__() 

124 

125 def __contains__(self, key: str) -> bool: 

126 return self._data.__contains__(key) 

127 

128 def keys(self) -> KeysView[str]: 

129 return self._data.keys() 

130 

131 def values(self) -> ValuesView[DataObjectList]: 

132 return self._data.values() 

133 

134 def items(self) -> ItemsView[str, DataObjectList]: 

135 return self._data.items() 

136 

137 def get(self, key: str, default: Any = None) -> DataObjectList | Any: 

138 return self._data.get(key, default) 

139 

140 def clear(self): 

141 return self._data.clear() 

142 

143 def setdefault(self, key: str, default: DataObjectList) -> DataObjectList: 

144 return self._data.setdefault(key, default) 

145 

146 def pop(self, key: str, default: Any = None) -> DataObjectList | Any: 

147 return self._data.pop(key, default) 

148 

149 def popitem(self) -> tuple[str, DataObjectList]: 

150 return self._data.popitem() 

151 

152 def copy(self) -> dict[str, DataObjectList]: 

153 return self._data.copy() 

154 

155 def deepcopy(self) -> dict[str, DataObjectList]: 

156 return deepcopy(self._data) 

157 

158 def update(self, **kwargs: DataObjectList): 

159 self._data.update(kwargs)