Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/bones/key.py: 62%

1import copy 

2import logging 

3import typing as t 

4from viur.core import db, i18n 

5from viur.core.bones.base import BaseBone, ReadFromClientError, ReadFromClientErrorSeverity 

6 

7 

8class KeyBone(BaseBone): 

9 """ 

10 The KeyBone is used for managing keys in the database. It provides various methods for validating, 

11 converting, and storing key values, as well as querying the database. 

12 Key management is crucial for maintaining relationships between entities in the database, and the 

13 KeyBone class helps ensure that keys are handled correctly and efficiently throughout the system. 

14 

15 :param descr: The description of the KeyBone. 

16 :param readOnly: Whether the KeyBone is read-only. 

17 :param visible: Whether the KeyBone is visible. 

18 :param allowed_kinds: The allowed entity kinds for the KeyBone. 

19 :param check: Whether to check for entity existence. 

20 """ 

21 type = "key" 

22 

23 def __init__( 

24 self, 

25 *, 

26 descr: str = "Key", 

27 readOnly: bool = True, # default is readonly 

28 visible: bool = False, # default is invisible 

29 allowed_kinds: t.Optional[t.Iterable[str]] = None, # None allows for any kind 

30 check: bool = False, # check for entity existence 

31 tags: str | t.Iterable[str] = "technical", 

32 **kwargs 

33 ): 

34 super().__init__(descr=descr, readOnly=readOnly, visible=visible, defaultValue=None, tags=tags, **kwargs) 

35 self.allowed_kinds = tuple(allowed_kinds) if allowed_kinds else None 

36 self.check = check 

37 

38 def singleValueFromClient(self, value, skel=None, bone_name=None, client_data=None, parse_only: bool = False): 

39 # check for correct key 

40 if isinstance(value, str): 

41 value = value.strip() 

42 

43 if self.allowed_kinds: 

44 try: 

45 key = db.key_helper(value, self.allowed_kinds[0], self.allowed_kinds[1:]) 

46 except ValueError as e: 

47 return self.getEmptyValue(), [ReadFromClientError(ReadFromClientErrorSeverity.Invalid, e.args[0])] 

48 else: 

49 try: 

50 key = db.normalize_key(value) 

51 except Exception as exc: 

52 logging.exception(f"Failed to normalize {value}: {exc}") 

53 return self.getEmptyValue(), [ 

54 ReadFromClientError( 

55 ReadFromClientErrorSeverity.Invalid, 

56 i18n.translate("core.bones.error.invalidkey", "No valid database key could be parsed") 

57 ) 

58 ] 

59 

60 if not parse_only: 60 ↛ 62line 60 didn't jump to line 62 because the condition on line 60 was never true

61 # Check custom validity 

62 if err := self.isInvalid(key): 

63 return self.getEmptyValue(), [ReadFromClientError(ReadFromClientErrorSeverity.Invalid, err)] 

64 

65 if self.check: 

66 if db.get(key) is None: 

67 return self.getEmptyValue(), [ 

68 ReadFromClientError( 

69 ReadFromClientErrorSeverity.Invalid, 

70 i18n.translate("core.bones.error.keynotfound", "The provided database key does not exist") 

71 ) 

72 ] 

73 

74 return key, None 

75 

76 def singleValueSerialize(self, value, skel: 'SkeletonInstance', name: str, parentIndexed: bool): 

77 if not value: 

78 return None 

79 

80 # Same parsing as reading from client: shortkeys via ``allowed_kinds``, keys normalized. 

81 key, err = self.singleValueFromClient(value, parse_only=True) 

82 if err: 

83 raise ValueError(err[0].errorMessage) 

84 

85 return key 

86 

87 def singleValueUnserialize(self, val): 

88 if not val: 

89 rval = None 

90 elif isinstance(val, db.Key): 

91 rval = db.normalize_key(val) 

92 else: 

93 rval, err = self.singleValueFromClient(val, parse_only=True) 

94 if err: 

95 raise ValueError(err[0].errorMessage) 

96 

97 return rval 

98 

99 def unserialize(self, skel: 'SkeletonInstance', name: str) -> bool: 

100 if ( 100 ↛ 108line 100 didn't jump to line 108 because the condition on line 100 was always true

101 name == "key" 

102 and isinstance(skel.dbEntity, db.Entity) 

103 and skel.dbEntity.key 

104 and not skel.dbEntity.key.is_partial 

105 ): 

106 skel.accessedValues[name] = skel.dbEntity.key 

107 return True 

108 return super().unserialize(skel, name) 

109 

110 def serialize(self, skel: 'SkeletonInstance', name: str, parentIndexed: bool) -> bool: 

111 if name == "key": 111 ↛ 112line 111 didn't jump to line 112 because the condition on line 111 was never true

112 if name not in skel.accessedValues: 

113 return False 

114 

115 skel.dbEntity.key = skel.accessedValues[name] 

116 return True 

117 

118 return super().serialize(skel, name, parentIndexed=parentIndexed) 

119 

120 def buildDBFilter( 

121 self, 

122 name: str, 

123 skel: 'viur.core.skeleton.SkeletonInstance', 

124 dbFilter: db.Query, 

125 rawFilter: dict, 

126 prefix: t.Optional[str] = None 

127 ) -> db.Query: 

128 """ 

129 This method parses the search filter specified by the client in their request and converts 

130 it into a format that can be understood by the datastore. It takes care of ignoring filters 

131 that do not target this bone and safely handles malformed data in the raw filter. 

132 

133 :param name: The property name of this bone in the Skeleton (not the description). 

134 :param skel: The :class:viur.core.skeleton.SkeletonInstance this bone is a part of. 

135 :param dbFilter: The current :class:viur.core.db.Query instance the filters should be 

136 applied to. 

137 :param rawFilter: The dictionary of filters the client wants to have applied. 

138 :param prefix: An optional string to prepend to the filter key. Defaults to None. 

139 

140 :return: The modified :class:viur.core.db.Query. 

141 

142 The method takes the following steps: 

143 

144 #. Decodes the provided key(s) from the raw filter. 

145 #. If the filter contains a list of keys, it iterates through the list, creating a new 

146 filter for each key and appending it to the list of queries. 

147 #. If the filter contains a single key, it applies the filter directly to the query. 

148 #. In case of any invalid key or other issues, it raises a RuntimeError. 

149 """ 

150 

151 def _decodeKey(key): 

152 if isinstance(key, db.Key): 

153 return key 

154 else: 

155 try: 

156 return db.Key.from_legacy_urlsafe(key) 

157 except Exception as e: 

158 logging.exception(e) 

159 logging.warning(f"Could not decode key {key}") 

160 raise RuntimeError() 

161 

162 if name in rawFilter: 162 ↛ exitline 162 didn't return from function 'buildDBFilter' because the condition on line 162 was always true

163 if isinstance(rawFilter[name], list): 

164 if isinstance(dbFilter.queries, list): 164 ↛ 165line 164 didn't jump to line 165 because the condition on line 164 was never true

165 raise ValueError("In-Filter already used!") 

166 elif dbFilter.queries is None: 166 ↛ 167line 166 didn't jump to line 167 because the condition on line 166 was never true

167 return dbFilter # Query is already unsatisfiable 

168 oldFilter = dbFilter.queries 

169 dbFilter.queries = [] 

170 for key in rawFilter[name]: 

171 newFilter = copy.deepcopy(oldFilter) 

172 try: 

173 if name == "key": 173 ↛ 176line 173 didn't jump to line 176 because the condition on line 173 was always true

174 newFilter.filters[f"{prefix or ''}{db.KEY_SPECIAL_PROPERTY} ="] = _decodeKey(key) 

175 else: 

176 newFilter.filters[f"{prefix or ''}{name} ="] = _decodeKey(key) 

177 except: # Invalid key or something 

178 raise RuntimeError() 

179 dbFilter.queries.append(newFilter) 

180 else: 

181 try: 

182 if name == "key": 

183 dbFilter.filter(f"""{prefix or ""}{db.KEY_SPECIAL_PROPERTY} =""", _decodeKey(rawFilter[name])) 

184 else: 

185 dbFilter.filter(f"""{prefix or ""}{name} =""", _decodeKey(rawFilter[name])) 

186 except: # Invalid key or something 

187 raise RuntimeError() 

188 return dbFilter 

189 

190 def _atomic_dump(self, value): 

191 if not value: 

192 return None 

193 

194 return str(value)