Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/utils/__init__.py: 40%

1import datetime 

2import logging 

3import typing as t 

4import urllib.parse 

5import warnings 

6import operator 

7from collections.abc import Iterable 

8from viur.core import current, db 

9from viur.core.config import conf 

10from deprecated.sphinx import deprecated 

11from . import json, parse, string # noqa: used by external imports 

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 viur.core.skeleton import SkeletonInstance 

15 

16 

17def utcNow() -> datetime.datetime: 

18 """ 

19 Returns an actual timestamp with UTC timezone setting. 

20 """ 

21 return datetime.datetime.now(datetime.timezone.utc) 

22 

23 

24def seoUrlToEntry(module: str, 

25 entry: t.Optional["SkeletonInstance"] = None, 

26 skelType: t.Optional[str] = None, 

27 language: t.Optional[str] = None) -> str: 

28 """ 

29 Return the seo-url to a skeleton instance or the module. 

30 

31 :param module: The module name. 

32 :param entry: A skeleton instance or None, to get the path to the module. 

33 :param skelType: # FIXME: Not used 

34 :param language: For which language. 

35 If None, the language of the current request is used. 

36 :return: The path (with a leading /). 

37 """ 

38 from viur.core import conf 

39 pathComponents = [""] 

40 if language is None: 

41 language = current.language.get() 

42 if conf.i18n.language_method == "url": 

43 pathComponents.append(language) 

44 if module in conf.i18n.language_module_map and language in conf.i18n.language_module_map[module]: 

45 module = conf.i18n.language_module_map[module][language] 

46 pathComponents.append(module) 

47 if not entry: 

48 return "/".join(pathComponents) 

49 else: 

50 try: 

51 currentSeoKeys = entry["viurCurrentSeoKeys"] 

52 except: 

53 return "/".join(pathComponents) 

54 if language in (currentSeoKeys or {}): 

55 pathComponents.append(str(currentSeoKeys[language])) 

56 elif "key" in entry: 

57 key = entry["key"] 

58 if isinstance(key, str): 

59 try: 

60 key = db.Key.from_legacy_urlsafe(key) 

61 except: 

62 pass 

63 pathComponents.append(str(key.id_or_name) if isinstance(key, db.Key) else str(key)) 

64 elif "name" in dir(entry): 

65 pathComponents.append(str(entry.name)) 

66 return "/".join(pathComponents) 

67 

68 

69def seoUrlToFunction(module: str, function: str, render: t.Optional[str] = None) -> str: 

70 from viur.core import conf 

71 lang = current.language.get() 

72 if module in conf.i18n.language_module_map and lang in conf.i18n.language_module_map[module]: 

73 module = conf.i18n.language_module_map[module][lang] 

74 if conf.i18n.language_method == "url": 

75 pathComponents = ["", lang] 

76 else: 

77 pathComponents = [""] 

78 targetObject = conf.main_resolver 

79 if module in targetObject: 

80 pathComponents.append(module) 

81 targetObject = targetObject[module] 

82 if render and render in targetObject: 

83 pathComponents.append(render) 

84 targetObject = targetObject[render] 

85 if function in targetObject: 

86 func = targetObject[function] 

87 if func.seo_language_map and lang in func.seo_language_map: 

88 pathComponents.append(func.seo_language_map[lang]) 

89 else: 

90 pathComponents.append(function) 

91 return "/".join(pathComponents) 

92 

93 

94@deprecated(version="3.8.0", reason="Use 'db.normalize_key' instead") 

95def normalizeKey(key: t.Union[None, "db.Key"]) -> t.Union[None, "db.Key"]: 

96 """ 

97 Normalizes a datastore key (replacing _application with the current one) 

98 

99 :param key: Key to be normalized. 

100 

101 :return: Normalized key in string representation. 

102 """ 

103 db.normalize_key(key) 

104 

105 

106def get_base_url() -> str: 

107 """ 

108 Retrieve current request's base URL with protocol. 

109 The function enforces use of https-protocol on non-localhost hostnames. 

110 

111 :returns: Returns the hostname, including the currently used protocol, e.g: https://www.example.com 

112 :rtype: str 

113 """ 

114 base = urllib.parse.urlparse(current.request.get().request.url).netloc # retrieve URL of request 

115 

116 # Always enforce https! 

117 if any(base.startswith(i) for i in ("localhost", "127.0.0.1", "[::1]", "0.0.0.0")): 

118 base = f"http://{base}" 

119 else: 

120 base = f"https://{base}" 

121 

122 # Replace non-SSL-ready-"appspot.com"-URLs with their SSL-ready counterpart 

123 return base.replace(f".{conf.instance.project_id}.", f"-dot-{conf.instance.project_id}.") 

124 

125 

126def ensure_iterable( 

127 obj: t.Any, 

128 *, 

129 test: t.Optional[t.Callable[[t.Any], bool]] = None, 

130 allow_callable: bool = True, 

131) -> t.Iterable[t.Any]: 

132 """ 

133 Ensures an object to be iterable. 

134 

135 An additional test can be provided to check additionally. 

136 

137 If the object is not considered to be iterable, a tuple with the object is returned. 

138 """ 

139 if allow_callable and callable(obj): 

140 obj = obj() 

141 

142 if not isinstance(obj, str) and isinstance(obj, Iterable): # uses collections.abc.Iterable 

143 if test is None or test(obj): 

144 return obj # return the obj, which is an iterable 

145 

146 return () # empty tuple 

147 

148 elif obj is None or (isinstance(obj, str) and not obj): 

149 return () # empty tuple 

150 

151 return obj, # return a tuple with the obj 

152 

153 

154def freeze_dict(value: dict[str, t.Any]) -> list: 

155 """Sort a dict recursively by keys and return as list""" 

156 return sorted( 

157 [ 

158 (pair[0], freeze_dict(pair[1])) if isinstance(pair[1], dict) else pair 

159 for pair in value.items() 

160 ], 

161 key=operator.itemgetter(0), 

162 ) 

163 

164 

165def build_content_disposition_header( 

166 filename: str, 

167 *, 

168 attachment: bool = False, 

169 inline: bool = False, 

170) -> str: 

171 """ 

172 Build a Content-Disposition header with UTF-8 support and ASCII fallback. 

173 

174 Generates a properly formatted `Content-Disposition` header value, including 

175 both a fallback ASCII filename and a UTF-8 encoded filename using RFC 5987. 

176 

177 Set either `attachment` or `inline` to control content disposition type. 

178 If both are False, the header will omit disposition type (not recommended). 

179 

180 Example: 

181 filename = "Änderung.pdf" ➜ 

182 'attachment; filename="Anderung.pdf"; filename*=UTF-8\'\'%C3%84nderung.pdf' 

183 

184 :param filename: The desired filename for the content. 

185 :param attachment: Whether to mark the content as an attachment. 

186 :param inline: Whether to mark the content as inline. 

187 :return: A `Content-Disposition` header string. 

188 """ 

189 if attachment and inline: 

190 raise ValueError("Only one of 'attachment' or 'inline' may be True.") 

191 

192 # Replace '+' with '%2B' in the ASCII fallback: when this header is embedded 

193 # as a query parameter in GCS signed URLs, the signing library leaves '+' unencoded 

194 # while GCS form-decodes '+' as space during verification → SignatureDoesNotMatch. 

195 fallback = string.normalize_ascii(filename).replace("+", "%2B") 

196 quoted_utf8 = urllib.parse.quote_from_bytes(filename.encode("utf-8")) 

197 

198 content_disposition = "; ".join( 

199 item for item in ( 

200 "attachment" if attachment else None, 

201 "inline" if inline else None, 

202 f'filename="{fallback}"' if filename else None, 

203 f'filename*=UTF-8\'\'{quoted_utf8}' if filename else None, 

204 ) if item 

205 ) 

206 

207 return content_disposition 

208 

209 

210# DEPRECATED ATTRIBUTES HANDLING 

211__UTILS_CONF_REPLACEMENT = { 

212 "projectID": "viur.instance.project_id", 

213 "isLocalDevelopmentServer": "viur.instance.is_dev_server", 

214 "projectBasePath": "viur.instance.project_base_path", 

215 "coreBasePath": "viur.instance.core_base_path" 

216} 

217 

218__UTILS_NAME_REPLACEMENT = { 

219 "currentLanguage": ("current.language", current.language), 

220 "currentRequest": ("current.request", current.request), 

221 "currentRequestData": ("current.request_data", current.request_data), 

222 "currentSession": ("current.session", current.session), 

223 "downloadUrlFor": ("conf.main_app.file.create_download_url", lambda: conf.main_app.file.create_download_url), 

224 "escapeString": ("utils.string.escape", lambda: string.escape), 

225 "generateRandomString": ("utils.string.random", lambda: string.random), 

226 "getCurrentUser": ("current.user.get", lambda: current.user.get), 

227 "is_prefix": ("utils.string.is_prefix", lambda: string.is_prefix), 

228 "parse_bool": ("utils.parse.bool", lambda: parse.bool), 

229 "srcSetFor": ("conf.main_app.file.create_src_set", lambda: conf.main_app.file.create_src_set), 

230} 

231 

232 

233def __getattr__(attr): 

234 if replace := __UTILS_CONF_REPLACEMENT.get(attr): 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true

235 msg = f"Use of `utils.{attr}` is deprecated; Use `conf.{replace}` instead!" 

236 warnings.warn(msg, DeprecationWarning, stacklevel=3) 

237 logging.warning(msg, stacklevel=3) 

238 return conf[replace] 

239 

240 if replace := __UTILS_NAME_REPLACEMENT.get(attr): 240 ↛ 241line 240 didn't jump to line 241 because the condition on line 240 was never true

241 msg = f"Use of `utils.{attr}` is deprecated; Use `{replace[0]}` instead!" 

242 warnings.warn(msg, DeprecationWarning, stacklevel=3) 

243 logging.warning(msg, stacklevel=3) 

244 res = replace[1] 

245 if isinstance(res, t.Callable): 

246 res = res() 

247 return res 

248 

249 return super(__import__(__name__).__class__).__getattribute__(attr)