Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/jinja2/bccache.py: 25%

1"""The optional bytecode cache system. This is useful if you have very 

2complex template situations and the compilation of all those templates 

3slows down your application too much. 

4 

5Situations where this is useful are often forking web applications that 

6are initialized on the first request. 

7""" 

8 

9import errno 

10import fnmatch 

11import marshal 

12import os 

13import pickle 

14import stat 

15import sys 

16import tempfile 

17import typing as t 

18from hashlib import sha1 

19from io import BytesIO 

20from types import CodeType 

21 

22if t.TYPE_CHECKING: 

23 import typing_extensions as te 

24 

25 from .environment import Environment 

26 

27 class _MemcachedClient(te.Protocol): 

28 def get(self, key: str) -> bytes: ... 

29 

30 def set( 

31 self, key: str, value: bytes, timeout: t.Optional[int] = None 

32 ) -> None: ... 

33 

34 

35bc_version = 5 

36# Magic bytes to identify Jinja bytecode cache files. Contains the 

37# Python major and minor version to avoid loading incompatible bytecode 

38# if a project upgrades its Python version. 

39bc_magic = ( 

40 b"j2" 

41 + pickle.dumps(bc_version, 2) 

42 + pickle.dumps((sys.version_info[0] << 24) | sys.version_info[1], 2) 

43) 

44 

45 

46class Bucket: 

47 """Buckets are used to store the bytecode for one template. It's created 

48 and initialized by the bytecode cache and passed to the loading functions. 

49 

50 The buckets get an internal checksum from the cache assigned and use this 

51 to automatically reject outdated cache material. Individual bytecode 

52 cache subclasses don't have to care about cache invalidation. 

53 """ 

54 

55 def __init__(self, environment: "Environment", key: str, checksum: str) -> None: 

56 self.environment = environment 

57 self.key = key 

58 self.checksum = checksum 

59 self.reset() 

60 

61 def reset(self) -> None: 

62 """Resets the bucket (unloads the bytecode).""" 

63 self.code: t.Optional[CodeType] = None 

64 

65 def load_bytecode(self, f: t.BinaryIO) -> None: 

66 """Loads bytecode from a file or file like object.""" 

67 # make sure the magic header is correct 

68 magic = f.read(len(bc_magic)) 

69 if magic != bc_magic: 

70 self.reset() 

71 return 

72 # the source code of the file changed, we need to reload 

73 checksum = pickle.load(f) 

74 if self.checksum != checksum: 

75 self.reset() 

76 return 

77 # if marshal_load fails then we need to reload 

78 try: 

79 self.code = marshal.load(f) 

80 except (EOFError, ValueError, TypeError): 

81 self.reset() 

82 return 

83 

84 def write_bytecode(self, f: t.IO[bytes]) -> None: 

85 """Dump the bytecode into the file or file like object passed.""" 

86 if self.code is None: 

87 raise TypeError("can't write empty bucket") 

88 f.write(bc_magic) 

89 pickle.dump(self.checksum, f, 2) 

90 marshal.dump(self.code, f) 

91 

92 def bytecode_from_string(self, string: bytes) -> None: 

93 """Load bytecode from bytes.""" 

94 self.load_bytecode(BytesIO(string)) 

95 

96 def bytecode_to_string(self) -> bytes: 

97 """Return the bytecode as bytes.""" 

98 out = BytesIO() 

99 self.write_bytecode(out) 

100 return out.getvalue() 

101 

102 

103class BytecodeCache: 

104 """To implement your own bytecode cache you have to subclass this class 

105 and override :meth:`load_bytecode` and :meth:`dump_bytecode`. Both of 

106 these methods are passed a :class:`~jinja2.bccache.Bucket`. 

107 

108 A very basic bytecode cache that saves the bytecode on the file system:: 

109 

110 from os import path 

111 

112 class MyCache(BytecodeCache): 

113 

114 def __init__(self, directory): 

115 self.directory = directory 

116 

117 def load_bytecode(self, bucket): 

118 filename = path.join(self.directory, bucket.key) 

119 if path.exists(filename): 

120 with open(filename, 'rb') as f: 

121 bucket.load_bytecode(f) 

122 

123 def dump_bytecode(self, bucket): 

124 filename = path.join(self.directory, bucket.key) 

125 with open(filename, 'wb') as f: 

126 bucket.write_bytecode(f) 

127 

128 A more advanced version of a filesystem based bytecode cache is part of 

129 Jinja. 

130 """ 

131 

132 def load_bytecode(self, bucket: Bucket) -> None: 

133 """Subclasses have to override this method to load bytecode into a 

134 bucket. If they are not able to find code in the cache for the 

135 bucket, it must not do anything. 

136 """ 

137 raise NotImplementedError() 

138 

139 def dump_bytecode(self, bucket: Bucket) -> None: 

140 """Subclasses have to override this method to write the bytecode 

141 from a bucket back to the cache. If it unable to do so it must not 

142 fail silently but raise an exception. 

143 """ 

144 raise NotImplementedError() 

145 

146 def clear(self) -> None: 

147 """Clears the cache. This method is not used by Jinja but should be 

148 implemented to allow applications to clear the bytecode cache used 

149 by a particular environment. 

150 """ 

151 

152 def get_cache_key( 

153 self, name: str, filename: t.Optional[t.Union[str]] = None 

154 ) -> str: 

155 """Returns the unique hash key for this template name.""" 

156 hash = sha1(name.encode("utf-8")) 

157 

158 if filename is not None: 

159 hash.update(f"|{filename}".encode()) 

160 

161 return hash.hexdigest() 

162 

163 def get_source_checksum(self, source: str) -> str: 

164 """Returns a checksum for the source.""" 

165 return sha1(source.encode("utf-8")).hexdigest() 

166 

167 def get_bucket( 

168 self, 

169 environment: "Environment", 

170 name: str, 

171 filename: t.Optional[str], 

172 source: str, 

173 ) -> Bucket: 

174 """Return a cache bucket for the given template. All arguments are 

175 mandatory but filename may be `None`. 

176 """ 

177 key = self.get_cache_key(name, filename) 

178 checksum = self.get_source_checksum(source) 

179 bucket = Bucket(environment, key, checksum) 

180 self.load_bytecode(bucket) 

181 return bucket 

182 

183 def set_bucket(self, bucket: Bucket) -> None: 

184 """Put the bucket into the cache.""" 

185 self.dump_bytecode(bucket) 

186 

187 

188class FileSystemBytecodeCache(BytecodeCache): 

189 """A bytecode cache that stores bytecode on the filesystem. It accepts 

190 two arguments: The directory where the cache items are stored and a 

191 pattern string that is used to build the filename. 

192 

193 If no directory is specified a default cache directory is selected. On 

194 Windows the user's temp directory is used, on UNIX systems a directory 

195 is created for the user in the system temp directory. 

196 

197 The pattern can be used to have multiple separate caches operate on the 

198 same directory. The default pattern is ``'__jinja2_%s.cache'``. ``%s`` 

199 is replaced with the cache key. 

200 

201 >>> bcc = FileSystemBytecodeCache('/tmp/jinja_cache', '%s.cache') 

202 

203 This bytecode cache supports clearing of the cache using the clear method. 

204 """ 

205 

206 def __init__( 

207 self, directory: t.Optional[str] = None, pattern: str = "__jinja2_%s.cache" 

208 ) -> None: 

209 if directory is None: 

210 directory = self._get_default_cache_dir() 

211 self.directory = directory 

212 self.pattern = pattern 

213 

214 def _get_default_cache_dir(self) -> str: 

215 def _unsafe_dir() -> "te.NoReturn": 

216 raise RuntimeError( 

217 "Cannot determine safe temp directory. You " 

218 "need to explicitly provide one." 

219 ) 

220 

221 tmpdir = tempfile.gettempdir() 

222 

223 # On windows the temporary directory is used specific unless 

224 # explicitly forced otherwise. We can just use that. 

225 if os.name == "nt": 

226 return tmpdir 

227 if not hasattr(os, "getuid"): 

228 _unsafe_dir() 

229 

230 dirname = f"_jinja2-cache-{os.getuid()}" 

231 actual_dir = os.path.join(tmpdir, dirname) 

232 

233 try: 

234 os.mkdir(actual_dir, stat.S_IRWXU) 

235 except OSError as e: 

236 if e.errno != errno.EEXIST: 

237 raise 

238 try: 

239 os.chmod(actual_dir, stat.S_IRWXU) 

240 actual_dir_stat = os.lstat(actual_dir) 

241 if ( 

242 actual_dir_stat.st_uid != os.getuid() 

243 or not stat.S_ISDIR(actual_dir_stat.st_mode) 

244 or stat.S_IMODE(actual_dir_stat.st_mode) != stat.S_IRWXU 

245 ): 

246 _unsafe_dir() 

247 except OSError as e: 

248 if e.errno != errno.EEXIST: 

249 raise 

250 

251 actual_dir_stat = os.lstat(actual_dir) 

252 if ( 

253 actual_dir_stat.st_uid != os.getuid() 

254 or not stat.S_ISDIR(actual_dir_stat.st_mode) 

255 or stat.S_IMODE(actual_dir_stat.st_mode) != stat.S_IRWXU 

256 ): 

257 _unsafe_dir() 

258 

259 return actual_dir 

260 

261 def _get_cache_filename(self, bucket: Bucket) -> str: 

262 return os.path.join(self.directory, self.pattern % (bucket.key,)) 

263 

264 def load_bytecode(self, bucket: Bucket) -> None: 

265 filename = self._get_cache_filename(bucket) 

266 

267 # Don't test for existence before opening the file, since the 

268 # file could disappear after the test before the open. 

269 try: 

270 f = open(filename, "rb") 

271 except (FileNotFoundError, IsADirectoryError, PermissionError): 

272 # PermissionError can occur on Windows when an operation is 

273 # in progress, such as calling clear(). 

274 return 

275 

276 with f: 

277 bucket.load_bytecode(f) 

278 

279 def dump_bytecode(self, bucket: Bucket) -> None: 

280 # Write to a temporary file, then rename to the real name after 

281 # writing. This avoids another process reading the file before 

282 # it is fully written. 

283 name = self._get_cache_filename(bucket) 

284 f = tempfile.NamedTemporaryFile( 

285 mode="wb", 

286 dir=os.path.dirname(name), 

287 prefix=os.path.basename(name), 

288 suffix=".tmp", 

289 delete=False, 

290 ) 

291 

292 def remove_silent() -> None: 

293 try: 

294 os.remove(f.name) 

295 except OSError: 

296 # Another process may have called clear(). On Windows, 

297 # another program may be holding the file open. 

298 pass 

299 

300 try: 

301 with f: 

302 bucket.write_bytecode(f) 

303 except BaseException: 

304 remove_silent() 

305 raise 

306 

307 try: 

308 os.replace(f.name, name) 

309 except OSError: 

310 # Another process may have called clear(). On Windows, 

311 # another program may be holding the file open. 

312 remove_silent() 

313 except BaseException: 

314 remove_silent() 

315 raise 

316 

317 def clear(self) -> None: 

318 # imported lazily here because google app-engine doesn't support 

319 # write access on the file system and the function does not exist 

320 # normally. 

321 from os import remove 

322 

323 files = fnmatch.filter(os.listdir(self.directory), self.pattern % ("*",)) 

324 for filename in files: 

325 try: 

326 remove(os.path.join(self.directory, filename)) 

327 except OSError: 

328 pass 

329 

330 

331class MemcachedBytecodeCache(BytecodeCache): 

332 """This class implements a bytecode cache that uses a memcache cache for 

333 storing the information. It does not enforce a specific memcache library 

334 (tummy's memcache or cmemcache) but will accept any class that provides 

335 the minimal interface required. 

336 

337 Libraries compatible with this class: 

338 

339 - `cachelib <https://github.com/pallets/cachelib>`_ 

340 - `python-memcached <https://pypi.org/project/python-memcached/>`_ 

341 

342 (Unfortunately the django cache interface is not compatible because it 

343 does not support storing binary data, only text. You can however pass 

344 the underlying cache client to the bytecode cache which is available 

345 as `django.core.cache.cache._client`.) 

346 

347 The minimal interface for the client passed to the constructor is this: 

348 

349 .. class:: MinimalClientInterface 

350 

351 .. method:: set(key, value[, timeout]) 

352 

353 Stores the bytecode in the cache. `value` is a string and 

354 `timeout` the timeout of the key. If timeout is not provided 

355 a default timeout or no timeout should be assumed, if it's 

356 provided it's an integer with the number of seconds the cache 

357 item should exist. 

358 

359 .. method:: get(key) 

360 

361 Returns the value for the cache key. If the item does not 

362 exist in the cache the return value must be `None`. 

363 

364 The other arguments to the constructor are the prefix for all keys that 

365 is added before the actual cache key and the timeout for the bytecode in 

366 the cache system. We recommend a high (or no) timeout. 

367 

368 This bytecode cache does not support clearing of used items in the cache. 

369 The clear method is a no-operation function. 

370 

371 .. versionadded:: 2.7 

372 Added support for ignoring memcache errors through the 

373 `ignore_memcache_errors` parameter. 

374 """ 

375 

376 def __init__( 

377 self, 

378 client: "_MemcachedClient", 

379 prefix: str = "jinja2/bytecode/", 

380 timeout: t.Optional[int] = None, 

381 ignore_memcache_errors: bool = True, 

382 ): 

383 self.client = client 

384 self.prefix = prefix 

385 self.timeout = timeout 

386 self.ignore_memcache_errors = ignore_memcache_errors 

387 

388 def load_bytecode(self, bucket: Bucket) -> None: 

389 try: 

390 code = self.client.get(self.prefix + bucket.key) 

391 except Exception: 

392 if not self.ignore_memcache_errors: 

393 raise 

394 else: 

395 bucket.bytecode_from_string(code) 

396 

397 def dump_bytecode(self, bucket: Bucket) -> None: 

398 key = self.prefix + bucket.key 

399 value = bucket.bytecode_to_string() 

400 

401 try: 

402 if self.timeout is not None: 

403 self.client.set(key, value, self.timeout) 

404 else: 

405 self.client.set(key, value) 

406 except Exception: 

407 if not self.ignore_memcache_errors: 

408 raise