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(self, key: str, value: bytes, timeout: int | None = None) -> None: ... 

31 

32 

33bc_version = 5 

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

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

36# if a project upgrades its Python version. 

37bc_magic = ( 

38 b"j2" 

39 + pickle.dumps(bc_version, 2) 

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

41) 

42 

43 

44class Bucket: 

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

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

47 

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

49 to automatically reject outdated cache material. Individual bytecode 

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

51 """ 

52 

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

54 self.environment = environment 

55 self.key = key 

56 self.checksum = checksum 

57 self.reset() 

58 

59 def reset(self) -> None: 

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

61 self.code: CodeType | None = None 

62 

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

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

65 # make sure the magic header is correct 

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

67 if magic != bc_magic: 

68 self.reset() 

69 return 

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

71 checksum = pickle.load(f) 

72 if self.checksum != checksum: 

73 self.reset() 

74 return 

75 # if marshal_load fails then we need to reload 

76 try: 

77 self.code = marshal.load(f) 

78 except (EOFError, ValueError, TypeError): 

79 self.reset() 

80 return 

81 

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

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

84 if self.code is None: 

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

86 f.write(bc_magic) 

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

88 marshal.dump(self.code, f) 

89 

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

91 """Load bytecode from bytes.""" 

92 self.load_bytecode(BytesIO(string)) 

93 

94 def bytecode_to_string(self) -> bytes: 

95 """Return the bytecode as bytes.""" 

96 out = BytesIO() 

97 self.write_bytecode(out) 

98 return out.getvalue() 

99 

100 

101class BytecodeCache: 

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

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

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

105 

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

107 

108 from os import path 

109 

110 class MyCache(BytecodeCache): 

111 

112 def __init__(self, directory): 

113 self.directory = directory 

114 

115 def load_bytecode(self, bucket): 

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

117 if path.exists(filename): 

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

119 bucket.load_bytecode(f) 

120 

121 def dump_bytecode(self, bucket): 

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

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

124 bucket.write_bytecode(f) 

125 

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

127 Jinja. 

128 """ 

129 

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

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

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

133 bucket, it must not do anything. 

134 """ 

135 raise NotImplementedError() 

136 

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

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

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

140 fail silently but raise an exception. 

141 """ 

142 raise NotImplementedError() 

143 

144 def clear(self) -> None: 

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

146 implemented to allow applications to clear the bytecode cache used 

147 by a particular environment. 

148 """ 

149 

150 def get_cache_key(self, name: str, filename: str | None = None) -> str: 

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

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

153 

154 if filename is not None: 

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

156 

157 return hash.hexdigest() 

158 

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

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

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

162 

163 def get_bucket( 

164 self, 

165 environment: "Environment", 

166 name: str, 

167 filename: str | None, 

168 source: str, 

169 ) -> Bucket: 

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

171 mandatory but filename may be `None`. 

172 """ 

173 key = self.get_cache_key(name, filename) 

174 checksum = self.get_source_checksum(source) 

175 bucket = Bucket(environment, key, checksum) 

176 self.load_bytecode(bucket) 

177 return bucket 

178 

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

180 """Put the bucket into the cache.""" 

181 self.dump_bytecode(bucket) 

182 

183 

184class FileSystemBytecodeCache(BytecodeCache): 

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

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

187 pattern string that is used to build the filename. 

188 

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

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

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

192 

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

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

195 is replaced with the cache key. 

196 

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

198 

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

200 """ 

201 

202 def __init__( 

203 self, directory: str | None = None, pattern: str = "__jinja2_%s.cache" 

204 ) -> None: 

205 if directory is None: 

206 directory = self._get_default_cache_dir() 

207 self.directory = directory 

208 self.pattern = pattern 

209 

210 def _get_default_cache_dir(self) -> str: 

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

212 raise RuntimeError( 

213 "Cannot determine safe temp directory. You " 

214 "need to explicitly provide one." 

215 ) 

216 

217 tmpdir = tempfile.gettempdir() 

218 

219 # On windows the temporary directory is used specific unless 

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

221 if os.name == "nt": 

222 return tmpdir 

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

224 _unsafe_dir() 

225 

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

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

228 

229 try: 

230 os.mkdir(actual_dir, stat.S_IRWXU) 

231 except OSError as e: 

232 if e.errno != errno.EEXIST: 

233 raise 

234 try: 

235 os.chmod(actual_dir, stat.S_IRWXU) 

236 actual_dir_stat = os.lstat(actual_dir) 

237 if ( 

238 actual_dir_stat.st_uid != os.getuid() 

239 or not stat.S_ISDIR(actual_dir_stat.st_mode) 

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

241 ): 

242 _unsafe_dir() 

243 except OSError as e: 

244 if e.errno != errno.EEXIST: 

245 raise 

246 

247 actual_dir_stat = os.lstat(actual_dir) 

248 if ( 

249 actual_dir_stat.st_uid != os.getuid() 

250 or not stat.S_ISDIR(actual_dir_stat.st_mode) 

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

252 ): 

253 _unsafe_dir() 

254 

255 return actual_dir 

256 

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

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

259 

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

261 filename = self._get_cache_filename(bucket) 

262 

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

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

265 try: 

266 f = open(filename, "rb") 

267 except (FileNotFoundError, IsADirectoryError, PermissionError): 

268 # PermissionError can occur on Windows when an operation is 

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

270 return 

271 

272 with f: 

273 bucket.load_bytecode(f) 

274 

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

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

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

278 # it is fully written. 

279 name = self._get_cache_filename(bucket) 

280 f = tempfile.NamedTemporaryFile( 

281 mode="wb", 

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

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

284 suffix=".tmp", 

285 delete=False, 

286 ) 

287 

288 def remove_silent() -> None: 

289 try: 

290 os.remove(f.name) 

291 except OSError: 

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

293 # another program may be holding the file open. 

294 pass 

295 

296 try: 

297 with f: 

298 bucket.write_bytecode(f) 

299 except BaseException: 

300 remove_silent() 

301 raise 

302 

303 try: 

304 os.replace(f.name, name) 

305 except OSError: 

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

307 # another program may be holding the file open. 

308 remove_silent() 

309 except BaseException: 

310 remove_silent() 

311 raise 

312 

313 def clear(self) -> None: 

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

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

316 # normally. 

317 from os import remove 

318 

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

320 for filename in files: 

321 try: 

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

323 except OSError: 

324 pass 

325 

326 

327class MemcachedBytecodeCache(BytecodeCache): 

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

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

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

331 the minimal interface required. 

332 

333 Libraries compatible with this class: 

334 

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

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

337 

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

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

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

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

342 

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

344 

345 .. class:: MinimalClientInterface 

346 

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

348 

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

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

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

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

353 item should exist. 

354 

355 .. method:: get(key) 

356 

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

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

359 

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

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

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

363 

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

365 The clear method is a no-operation function. 

366 

367 .. versionadded:: 2.7 

368 Added support for ignoring memcache errors through the 

369 `ignore_memcache_errors` parameter. 

370 """ 

371 

372 def __init__( 

373 self, 

374 client: "_MemcachedClient", 

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

376 timeout: int | None = None, 

377 ignore_memcache_errors: bool = True, 

378 ): 

379 self.client = client 

380 self.prefix = prefix 

381 self.timeout = timeout 

382 self.ignore_memcache_errors = ignore_memcache_errors 

383 

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

385 try: 

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

387 except Exception: 

388 if not self.ignore_memcache_errors: 

389 raise 

390 else: 

391 bucket.bytecode_from_string(code) 

392 

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

394 key = self.prefix + bucket.key 

395 value = bucket.bytecode_to_string() 

396 

397 try: 

398 if self.timeout is not None: 

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

400 else: 

401 self.client.set(key, value) 

402 except Exception: 

403 if not self.ignore_memcache_errors: 

404 raise