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

1import datetime 

2import logging 

3import typing as t 

4import urllib.parse 

5import warnings 

6from collections.abc import Iterable 

7 

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 

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

111 :rtype: str 

112 """ 

113 url = current.request.get().request.url # retrireve URL by request 

114 url = url[:url.find("/", url.find("://") + 5)] # cut out base-url 

115 

116 # Always enforce https! 

117 if not any(url.startswith(f"http://{i}") for i in ("localhost", "127.0.0.1")): 

118 url = "https://" + url[7:] 

119 

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

121 return url.replace(f".{conf.instance.project_id}.", f"-dot-{conf.instance.project_id}.") 

122 

123 

124def ensure_iterable( 

125 obj: t.Any, 

126 *, 

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

128 allow_callable: bool = True, 

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

130 """ 

131 Ensures an object to be iterable. 

132 

133 An additional test can be provided to check additionally. 

134 

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

136 """ 

137 if allow_callable and callable(obj): 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true

138 obj = obj() 

139 

140 if not isinstance(obj, str) and isinstance(obj, Iterable): # uses collections.abc.Iterable 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true

141 if test is None or test(obj): 

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

143 

144 return () # empty tuple 

145 

146 elif obj is None or (isinstance(obj, str) and not obj): 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true

147 return () # empty tuple 

148 

149 return obj, # return a tuple with the obj 

150 

151 

152def build_content_disposition_header( 

153 filename: str, 

154 *, 

155 attachment: bool = False, 

156 inline: bool = False, 

157) -> str: 

158 """ 

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

160 

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

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

163 

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

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

166 

167 Example: 

168 filename = "Änderung.pdf" ➜ 

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

170 

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

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

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

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

175 """ 

176 if attachment and inline: 

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

178 

179 fallback = string.normalize_ascii(filename) 

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

181 

182 content_disposition = "; ".join( 

183 item for item in ( 

184 "attachment" if attachment else None, 

185 "inline" if inline else None, 

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

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

188 ) if item 

189 ) 

190 

191 return content_disposition 

192 

193 

194# DEPRECATED ATTRIBUTES HANDLING 

195__UTILS_CONF_REPLACEMENT = { 

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

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

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

199 "coreBasePath": "viur.instance.core_base_path" 

200} 

201 

202__UTILS_NAME_REPLACEMENT = { 

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

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

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

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

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

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

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

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

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

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

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

214} 

215 

216 

217def __getattr__(attr): 

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

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

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

221 logging.warning(msg, stacklevel=3) 

222 return conf[replace] 

223 

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

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

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

227 logging.warning(msg, stacklevel=3) 

228 res = replace[1] 

229 if isinstance(res, t.Callable): 

230 res = res() 

231 return res 

232 

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