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

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: 

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 singleValueUnserialize(self, val): 

77 if not val: 

78 rval = None 

79 elif isinstance(val, db.Key): 

80 rval = db.normalize_key(val) 

81 else: 

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

83 if err: 

84 raise ValueError(err[0].errorMessage) 

85 

86 return rval 

87 

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

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

90 name == "key" 

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

92 and skel.dbEntity.key 

93 and not skel.dbEntity.key.is_partial 

94 ): 

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

96 return True 

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

98 

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

100 if name == "key": 

101 if name not in skel.accessedValues: 

102 return False 

103 

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

105 return True 

106 

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

108 

109 def buildDBFilter( 

110 self, 

111 name: str, 

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

113 dbFilter: db.Query, 

114 rawFilter: dict, 

115 prefix: t.Optional[str] = None 

116 ) -> db.Query: 

117 """ 

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

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

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

121 

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

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

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

125 applied to. 

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

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

128 

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

130 

131 The method takes the following steps: 

132 

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

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

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

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

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

138 """ 

139 

140 def _decodeKey(key): 

141 if isinstance(key, db.Key): 

142 return key 

143 else: 

144 try: 

145 return db.Key.from_legacy_urlsafe(key) 

146 except Exception as e: 

147 logging.exception(e) 

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

149 raise RuntimeError() 

150 

151 if name in rawFilter: 

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

153 if isinstance(dbFilter.queries, list): 

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

155 elif dbFilter.queries is None: 

156 return dbFilter # Query is already unsatisfiable 

157 oldFilter = dbFilter.queries 

158 dbFilter.queries = [] 

159 for key in rawFilter[name]: 

160 newFilter = copy.deepcopy(oldFilter) 

161 try: 

162 if name == "key": 

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

164 else: 

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

166 except: # Invalid key or something 

167 raise RuntimeError() 

168 dbFilter.queries.append(newFilter) 

169 else: 

170 try: 

171 if name == "key": 

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

173 else: 

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

175 except: # Invalid key or something 

176 raise RuntimeError() 

177 return dbFilter 

178 

179 def _atomic_dump(self, value): 

180 if not value: 

181 return None 

182 

183 return str(value)