Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/mako/util.py: 24%

1# mako/util.py 

2# Copyright 2006-2026 the Mako authors and contributors <see AUTHORS file> 

3# 

4# This module is part of Mako and is released under 

5# the MIT License: http://www.opensource.org/licenses/mit-license.php 

6from ast import parse 

7import codecs 

8import collections 

9import operator 

10import os 

11import re 

12import timeit 

13 

14from .compat import importlib_metadata_get 

15 

16 

17def update_wrapper(decorated, fn): 

18 decorated.__wrapped__ = fn 

19 decorated.__name__ = fn.__name__ 

20 return decorated 

21 

22 

23class PluginLoader: 

24 def __init__(self, group): 

25 self.group = group 

26 self.impls = {} 

27 

28 def load(self, name): 

29 if name in self.impls: 

30 return self.impls[name]() 

31 

32 for impl in importlib_metadata_get(self.group): 

33 if impl.name == name: 

34 self.impls[name] = impl.load 

35 return impl.load() 

36 

37 from mako import exceptions 

38 

39 raise exceptions.RuntimeException( 

40 "Can't load plugin %s %s" % (self.group, name) 

41 ) 

42 

43 def register(self, name, modulepath, objname): 

44 def load(): 

45 mod = __import__(modulepath) 

46 for token in modulepath.split(".")[1:]: 

47 mod = getattr(mod, token) 

48 return getattr(mod, objname) 

49 

50 self.impls[name] = load 

51 

52 

53def verify_directory(dir_): 

54 """create and/or verify a filesystem directory.""" 

55 

56 tries = 0 

57 

58 while not os.path.exists(dir_): 

59 try: 

60 tries += 1 

61 os.makedirs(dir_, 0o755) 

62 except: 

63 if tries > 5: 

64 raise 

65 

66 

67def to_list(x, default=None): 

68 if x is None: 

69 return default 

70 if not isinstance(x, (list, tuple)): 

71 return [x] 

72 else: 

73 return x 

74 

75 

76class memoized_property: 

77 """A read-only @property that is only evaluated once.""" 

78 

79 def __init__(self, fget, doc=None): 

80 self.fget = fget 

81 self.__doc__ = doc or fget.__doc__ 

82 self.__name__ = fget.__name__ 

83 

84 def __get__(self, obj, cls): 

85 if obj is None: 

86 return self 

87 obj.__dict__[self.__name__] = result = self.fget(obj) 

88 return result 

89 

90 

91class memoized_instancemethod: 

92 """Decorate a method memoize its return value. 

93 

94 Best applied to no-arg methods: memoization is not sensitive to 

95 argument values, and will always return the same value even when 

96 called with different arguments. 

97 

98 """ 

99 

100 def __init__(self, fget, doc=None): 

101 self.fget = fget 

102 self.__doc__ = doc or fget.__doc__ 

103 self.__name__ = fget.__name__ 

104 

105 def __get__(self, obj, cls): 

106 if obj is None: 

107 return self 

108 

109 def oneshot(*args, **kw): 

110 result = self.fget(obj, *args, **kw) 

111 

112 def memo(*a, **kw): 

113 return result 

114 

115 memo.__name__ = self.__name__ 

116 memo.__doc__ = self.__doc__ 

117 obj.__dict__[self.__name__] = memo 

118 return result 

119 

120 oneshot.__name__ = self.__name__ 

121 oneshot.__doc__ = self.__doc__ 

122 return oneshot 

123 

124 

125class SetLikeDict(dict): 

126 """a dictionary that has some setlike methods on it""" 

127 

128 def union(self, other): 

129 """produce a 'union' of this dict and another (at the key level). 

130 

131 values in the second dict take precedence over that of the first""" 

132 x = SetLikeDict(**self) 

133 x.update(other) 

134 return x 

135 

136 

137class FastEncodingBuffer: 

138 """a very rudimentary buffer that is faster than StringIO, 

139 and supports unicode data.""" 

140 

141 def __init__(self, encoding=None, errors="strict"): 

142 self.data = collections.deque() 

143 self.encoding = encoding 

144 self.delim = "" 

145 self.errors = errors 

146 self.write = self.data.append 

147 

148 def truncate(self): 

149 self.data = collections.deque() 

150 self.write = self.data.append 

151 

152 def getvalue(self): 

153 if self.encoding: 

154 return self.delim.join(self.data).encode( 

155 self.encoding, self.errors 

156 ) 

157 else: 

158 return self.delim.join(self.data) 

159 

160 

161class LRUCache(dict): 

162 """A dictionary-like object that stores a limited number of items, 

163 discarding lesser used items periodically. 

164 

165 this is a rewrite of LRUCache from Myghty to use a periodic timestamp-based 

166 paradigm so that synchronization is not really needed. the size management 

167 is inexact. 

168 """ 

169 

170 class _Item: 

171 def __init__(self, key, value): 

172 self.key = key 

173 self.value = value 

174 self.timestamp = timeit.default_timer() 

175 

176 def __repr__(self): 

177 return repr(self.value) 

178 

179 def __init__(self, capacity, threshold=0.5): 

180 self.capacity = capacity 

181 self.threshold = threshold 

182 

183 def __getitem__(self, key): 

184 item = dict.__getitem__(self, key) 

185 item.timestamp = timeit.default_timer() 

186 return item.value 

187 

188 def values(self): 

189 return [i.value for i in dict.values(self)] 

190 

191 def setdefault(self, key, value): 

192 if key in self: 

193 return self[key] 

194 self[key] = value 

195 return value 

196 

197 def __setitem__(self, key, value): 

198 item = dict.get(self, key) 

199 if item is None: 

200 item = self._Item(key, value) 

201 dict.__setitem__(self, key, item) 

202 else: 

203 item.value = value 

204 self._manage_size() 

205 

206 def _manage_size(self): 

207 while len(self) > self.capacity + self.capacity * self.threshold: 

208 bytime = sorted( 

209 dict.values(self), 

210 key=operator.attrgetter("timestamp"), 

211 reverse=True, 

212 ) 

213 for item in bytime[self.capacity :]: 

214 try: 

215 del self[item.key] 

216 except KeyError: 

217 # if we couldn't find a key, most likely some other thread 

218 # broke in on us. loop around and try again 

219 break 

220 

221 

222# Regexp to match python magic encoding line 

223_PYTHON_MAGIC_COMMENT_re = re.compile( 

224 r"[ \t\f]* \# .* coding[=:][ \t]*([-\w.]+)", re.VERBOSE 

225) 

226 

227 

228def parse_encoding(fp): 

229 """Deduce the encoding of a Python source file (binary mode) from magic 

230 comment. 

231 

232 It does this in the same way as the `Python interpreter`__ 

233 

234 .. __: http://docs.python.org/ref/encodings.html 

235 

236 The ``fp`` argument should be a seekable file object in binary mode. 

237 """ 

238 pos = fp.tell() 

239 fp.seek(0) 

240 try: 

241 line1 = fp.readline() 

242 has_bom = line1.startswith(codecs.BOM_UTF8) 

243 if has_bom: 

244 line1 = line1[len(codecs.BOM_UTF8) :] 

245 

246 m = _PYTHON_MAGIC_COMMENT_re.match(line1.decode("ascii", "ignore")) 

247 if not m: 

248 try: 

249 parse(line1.decode("ascii", "ignore")) 

250 except (ImportError, SyntaxError): 

251 # Either it's a real syntax error, in which case the source 

252 # is not valid python source, or line2 is a continuation of 

253 # line1, in which case we don't want to scan line2 for a magic 

254 # comment. 

255 pass 

256 else: 

257 line2 = fp.readline() 

258 m = _PYTHON_MAGIC_COMMENT_re.match( 

259 line2.decode("ascii", "ignore") 

260 ) 

261 

262 if has_bom: 

263 if m: 

264 raise SyntaxError( 

265 "python refuses to compile code with both a UTF8" 

266 " byte-order-mark and a magic encoding comment" 

267 ) 

268 return "utf_8" 

269 elif m: 

270 return m.group(1) 

271 else: 

272 return None 

273 finally: 

274 fp.seek(pos) 

275 

276 

277def sorted_dict_repr(d): 

278 """repr() a dictionary with the keys in order. 

279 

280 Used by the lexer unit test to compare parse trees based on strings. 

281 

282 """ 

283 keys = list(d.keys()) 

284 keys.sort() 

285 return "{" + ", ".join("%r: %r" % (k, d[k]) for k in keys) + "}" 

286 

287 

288def read_file(path, mode="rb"): 

289 with open(path, mode) as fp: 

290 return fp.read() 

291 

292 

293def read_python_file(path): 

294 fp = open(path, "rb") 

295 try: 

296 encoding = parse_encoding(fp) 

297 data = fp.read() 

298 if encoding: 

299 data = data.decode(encoding) 

300 return data 

301 finally: 

302 fp.close()