Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/bones/string.py: 56%

159 statements  

« prev     ^ index     » next       coverage.py v7.16.1, created at 2026-09-25 15:02 +0000

1import datetime 

2import functools 

3import logging 

4import string 

5import typing as t 

6import warnings 

7from numbers import Number 

8 

9from viur.core import conf, current, db, utils 

10from .base import ReadFromClientError, ReadFromClientErrorSeverity 

11from .raw import RawBone 

12 

13if t.TYPE_CHECKING: 13 ↛ 14line 13 didn't jump to line 14 because the condition on line 13 was never true

14 from ..skeleton import SkeletonInstance 

15 

16DB_TYPE_INDEXED: t.TypeAlias = dict[t.Literal["val", "idx", "sort_idx"], str] 

17 

18 

19class StringBone(RawBone): 

20 """ 

21 The "StringBone" represents a data field that contains text values. 

22 """ 

23 type = "str" 

24 

25 def __init__( 

26 self, 

27 *, 

28 caseSensitive: bool = True, 

29 max_length: int | None = 254, 

30 min_length: int | None = None, 

31 natural_sorting: bool | t.Callable = False, 

32 escape_html: bool | None = None, 

33 **kwargs 

34 ): 

35 """ 

36 Initializes a new StringBone. 

37 

38 :param caseSensitive: When filtering for values in this bone, should it be case-sensitive? 

39 :param max_length: The maximum length allowed for values of this bone. Set to None for no limitation. 

40 :param min_length: The minimum length allowed for values of this bone. Set to None for no limitation. 

41 :param natural_sorting: Allows a more natural sorting 

42 than the default sorting on the plain values. 

43 This uses the .sort_idx property. 

44 `True` enables sorting according to DIN 5007 Variant 2. 

45 With passing a `callable`, a custom transformer method can be set 

46 that creates the value for the index property. 

47 :param escape_html: Replace some characters in the string with HTML-safe sequences with 

48 using :meth:`utils.string.escape` for safe use in HTML. 

49 Defaults to :attr:`conf.bone_string_escape_html` if not set explicitly. 

50 :param kwargs: Inherited arguments from the BaseBone. 

51 """ 

52 # fixme: Remove in viur-core >= 4 

53 if "maxLength" in kwargs: 53 ↛ 54line 53 didn't jump to line 54 because the condition on line 53 was never true

54 warnings.warn("maxLength parameter is deprecated, please use max_length", 

55 DeprecationWarning, stacklevel=2) 

56 max_length = kwargs.pop("maxLength") 

57 super().__init__(**kwargs) 

58 if max_length is not None and max_length <= 0: 

59 raise ValueError("max_length must be a positive integer or None") 

60 if min_length is not None and min_length <= 0: 

61 raise ValueError("min_length must be a positive integer or None") 

62 if min_length is not None and max_length is not None: 

63 if min_length > max_length: 

64 raise ValueError("min_length can't be greater than max_length") 

65 self.caseSensitive = caseSensitive 

66 self.max_length = max_length 

67 self.min_length = min_length 

68 if callable(natural_sorting): 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true

69 self.natural_sorting = natural_sorting 

70 elif not isinstance(natural_sorting, bool): 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true

71 raise TypeError("natural_sorting must be a callable or boolean!") 

72 elif not natural_sorting: 

73 self.natural_sorting = None 

74 # else: keep self.natural_sorting as is 

75 self.escape_html = conf.bone_string_escape_html if escape_html is None else escape_html 

76 

77 def type_coerce_single_value(self, value: t.Any) -> str: 

78 """Convert a value to a string (if not already) 

79 

80 Converts a value that is not a string into a string 

81 if a meaningful conversion is possible (simple data types only). 

82 """ 

83 if isinstance(value, str): 

84 return value 

85 elif isinstance(value, Number): 

86 return str(value) 

87 elif isinstance(value, (datetime.datetime, datetime.date, datetime.time)): 

88 return value.isoformat() 

89 elif isinstance(value, db.Key): 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true

90 return value.to_legacy_urlsafe().decode("ASCII") 

91 elif not value: # None or any other falsy value 

92 return self.getEmptyValue() 

93 else: 

94 raise ValueError( 

95 f"Value {value} of type {type(value)} cannot be coerced for {type(self).__name__} {self.name}" 

96 ) 

97 

98 def singleValueSerialize( 

99 self, 

100 value: t.Any, 

101 skel: "SkeletonInstance", 

102 name: str, 

103 parentIndexed: bool, 

104 ) -> str | DB_TYPE_INDEXED: 

105 """ 

106 Serializes a single value of this data field for storage in the database. 

107 

108 :param value: The value to serialize. 

109 It should be a str value, if not it is forced with :meth:`type_coerce_single_value`. 

110 :param skel: The skeleton instance that this data field belongs to. 

111 :param name: The name of this data field. 

112 :param parentIndexed: A boolean value indicating whether the parent object has an index on 

113 this data field or not. 

114 :return: The serialized value. 

115 """ 

116 value = self.type_coerce_single_value(value) 

117 if (not self.caseSensitive or self.natural_sorting) and parentIndexed: 

118 serialized: DB_TYPE_INDEXED = {"val": value} 

119 if not self.caseSensitive: 119 ↛ 121line 119 didn't jump to line 121 because the condition on line 119 was always true

120 serialized["idx"] = value.lower() 

121 if self.natural_sorting: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true

122 serialized["sort_idx"] = self.natural_sorting(value) 

123 return serialized 

124 return value 

125 

126 def singleValueUnserialize(self, value: str | DB_TYPE_INDEXED) -> str: 

127 """ 

128 Unserializes a single value of this data field from the database. 

129 

130 :param value: The serialized value to unserialize. 

131 :return: The unserialized value. 

132 """ 

133 if isinstance(value, dict) and "val" in value: 

134 value = value["val"] # Process with the raw value 

135 if value: 

136 return str(value) 

137 else: 

138 return self.getEmptyValue() 

139 

140 def getEmptyValue(self) -> str: 

141 """ 

142 Returns the empty value for this data field. 

143 

144 :return: An empty string. 

145 """ 

146 return "" 

147 

148 def isEmpty(self, value): 

149 """ 

150 Determines whether a value for this data field is empty or not. 

151 

152 :param value: The value to check for emptiness. 

153 :return: A boolean value indicating whether the value is empty or not. 

154 """ 

155 if not value: 

156 return True 

157 

158 return not bool(str(value).strip()) 

159 

160 def isInvalid(self, value: t.Any) -> str | None: 

161 """ 

162 Returns None if the value would be valid for 

163 this bone, an error-message otherwise. 

164 """ 

165 if self.max_length is not None and len(value) > self.max_length: 

166 return "Maximum length exceeded" 

167 if self.min_length is not None and len(value) < self.min_length: 

168 return "Minimum length not reached" 

169 return None 

170 

171 def singleValueFromClient(self, value, skel, bone_name, client_data): 

172 """ 

173 Returns None and the escaped value if the value would be valid for 

174 this bone, otherwise the empty value and an error-message. 

175 """ 

176 if not (err := self.isInvalid(str(value))): 

177 if self.escape_html: 

178 return utils.string.escape(value, self.max_length), None 

179 elif self.max_length: 179 ↛ 181line 179 didn't jump to line 181 because the condition on line 179 was always true

180 return value[:self.max_length], None 

181 return value, None 

182 

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

184 

185 def buildDBFilter( 

186 self, 

187 name: str, 

188 skel: "SkeletonInstance", 

189 dbFilter: db.Query, 

190 rawFilter: dict, 

191 prefix: t.Optional[str] = None 

192 ) -> db.Query: 

193 """ 

194 Builds and returns a database filter for this data field based on the provided raw filter data. 

195 

196 :param name: The name of this data field. 

197 :param skel: The skeleton instance that this data field belongs to. 

198 :param dbFilter: The database filter to add query clauses to. 

199 :param rawFilter: A dictionary containing the raw filter data for this data field. 

200 :param prefix: An optional prefix to add to the query clause. 

201 :return: The database filter with the added query clauses. 

202 """ 

203 if name not in rawFilter and not any( 

204 [(x.startswith(name + "$") or x.startswith(name + ".")) for x in rawFilter.keys()] 

205 ): 

206 return super().buildDBFilter(name, skel, dbFilter, rawFilter, prefix) 

207 

208 if not self.languages: 

209 namefilter = name 

210 else: 

211 lang = None 

212 for key in rawFilter.keys(): 

213 if key.startswith(f"{name}."): 

214 langStr = key.replace(f"{name}.", "") 

215 if langStr in self.languages: 

216 lang = langStr 

217 break 

218 if not lang: 

219 lang = current.language.get() # currentSession.getLanguage() 

220 if not lang or not lang in self.languages: 

221 lang = self.languages[0] 

222 namefilter = f"{name}.{lang}" 

223 

224 if name + "$lk" in rawFilter: # Do a prefix-match 

225 if not self.caseSensitive: 

226 dbFilter.filter((prefix or "") + namefilter + ".idx >=", str(rawFilter[name + "$lk"]).lower()) 

227 dbFilter.filter((prefix or "") + namefilter + ".idx <", 

228 str(rawFilter[name + "$lk"] + u"\ufffd").lower()) 

229 else: 

230 dbFilter.filter((prefix or "") + namefilter + " >=", str(rawFilter[name + "$lk"])) 

231 dbFilter.filter((prefix or "") + namefilter + " <", str(rawFilter[name + "$lk"] + u"\ufffd")) 

232 

233 if name + "$gt" in rawFilter: # All entries after 

234 if not self.caseSensitive: 

235 dbFilter.filter((prefix or "") + namefilter + ".idx >", str(rawFilter[name + "$gt"]).lower()) 

236 else: 

237 dbFilter.filter((prefix or "") + namefilter + " >", str(rawFilter[name + "$gt"])) 

238 

239 if name + "$lt" in rawFilter: # All entries before 

240 if not self.caseSensitive: 

241 dbFilter.filter((prefix or "") + namefilter + ".idx <", str(rawFilter[name + "$lt"]).lower()) 

242 else: 

243 dbFilter.filter((prefix or "") + namefilter + " <", str(rawFilter[name + "$lt"])) 

244 

245 if name in rawFilter: # Normal, strict match 

246 if not self.caseSensitive: 

247 dbFilter.filter((prefix or "") + namefilter + ".idx", str(rawFilter[name]).lower()) 

248 else: 

249 dbFilter.filter((prefix or "") + namefilter, str(rawFilter[name])) 

250 

251 return dbFilter 

252 

253 def buildDBSort( 

254 self, 

255 name: str, 

256 skel: 'SkeletonInstance', 

257 query: db.Query, 

258 params: dict, 

259 postfix: str = "", 

260 ) -> t.Optional[db.Query]: 

261 return super().buildDBSort( 

262 name, skel, query, params, 

263 postfix=".sort_idx" if self.natural_sorting else ".idx" if not self.caseSensitive else postfix 

264 ) 

265 

266 def natural_sorting(self, value: str | None) -> str | None: 

267 """Implements a default natural sorting transformer. 

268 

269 The sorting is according to DIN 5007 Variant 2 

270 and sets ö and oe, etc. equal. 

271 """ 

272 if value is None: 

273 return None 

274 assert isinstance(value, str) 

275 if not self.caseSensitive: 

276 value = value.lower() 

277 

278 # DIN 5007 Variant 2 

279 return value.translate(str.maketrans({ 

280 "ö": "oe", 

281 "Ö": "Oe", 

282 "ü": "ue", 

283 "Ü": "Ue", 

284 "ä": "ae", 

285 "Ä": "Ae", 

286 "ß": "ss", 

287 "ẞ": "SS", 

288 })) 

289 

290 def getUniquePropertyIndexValues(self, skel: "SkeletonInstance", name: str) -> list[str]: 

291 """ 

292 Returns a list of unique index values for a given property name. 

293 

294 :param skel: The skeleton instance. 

295 :param name: The name of the property. 

296 :return: A list of unique index values for the property. 

297 """ 

298 if not self.caseSensitive: 

299 values = [ 

300 value.lower() if isinstance(value, str) else value 

301 for _, _, value in self.iter_bone_value(skel, name) 

302 if value is not None 

303 ] 

304 

305 return self._hashValueForUniquePropertyIndex(values) if values else [] 

306 

307 return super().getUniquePropertyIndexValues(skel, name) 

308 

309 def refresh(self, skel: "SkeletonInstance", bone_name: str) -> None: 

310 super().refresh(skel, bone_name) 

311 

312 # TODO: duplicate code, this is the same iteration logic as in NumericBone 

313 new_value = {} 

314 for _, lang, value in self.iter_bone_value(skel, bone_name): 

315 value = self.type_coerce_single_value(value) 

316 if self.escape_html: 

317 value = utils.string.escape(value) 

318 else: 

319 value = utils.string.unescape(value) 

320 new_value.setdefault(lang, []).append(value) 

321 

322 if not self.multiple: 

323 # take the first one 

324 new_value = {lang: values[0] for lang, values in new_value.items() if values} 

325 

326 if self.languages: 

327 skel[bone_name] = new_value 

328 elif not self.languages: 

329 # just the value(s) with None language 

330 skel[bone_name] = new_value.get(None, [] if self.multiple else self.getEmptyValue()) 

331 

332 def structure(self) -> dict: 

333 ret = super().structure() | { 

334 "maxlength": self.max_length, 

335 "minlength": self.min_length 

336 } 

337 return ret 

338 

339 @classmethod 

340 def v_func_valid_chars(cls, valid_chars: t.Iterable = string.printable) -> t.Callable: 

341 """ 

342 Returns a function that takes a string and checks whether it contains valid characters. 

343 If all characters of the string are valid, it returns None, and succeeds. 

344 If invalid characters are present, it returns an appropriate error message. 

345 

346 :param valid_chars: An iterable of valid characters. 

347 :return: A function that takes a string and check whether it contains valid characters. 

348 

349 Example for digits only: 

350 .. code-block:: python 

351 str_bone = StringBone(vfunc=StringBone.v_func_valid_chars(string.digits)) 

352 """ 

353 

354 def v_func(valid_chars_intern, value): 

355 if any(char not in valid_chars_intern for char in value): 

356 return "Not all letters are available in the charset" 

357 

358 return functools.partial(v_func, valid_chars)