jinja2.bccache
The optional bytecode cache system. This is useful if you have very complex template situations and the compilation of all those templates slows down your application too much.
Situations where this is useful are often forking web applications that are initialized on the first request.
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
bc_version =
5
bc_magic =
b'j2\x80\x02K\x05.\x80\x02J\x0c\x00\x00\x03.'
class
Bucket:
47class Bucket: 48 """Buckets are used to store the bytecode for one template. It's created 49 and initialized by the bytecode cache and passed to the loading functions. 50 51 The buckets get an internal checksum from the cache assigned and use this 52 to automatically reject outdated cache material. Individual bytecode 53 cache subclasses don't have to care about cache invalidation. 54 """ 55 56 def __init__(self, environment: "Environment", key: str, checksum: str) -> None: 57 self.environment = environment 58 self.key = key 59 self.checksum = checksum 60 self.reset() 61 62 def reset(self) -> None: 63 """Resets the bucket (unloads the bytecode).""" 64 self.code: t.Optional[CodeType] = None 65 66 def load_bytecode(self, f: t.BinaryIO) -> None: 67 """Loads bytecode from a file or file like object.""" 68 # make sure the magic header is correct 69 magic = f.read(len(bc_magic)) 70 if magic != bc_magic: 71 self.reset() 72 return 73 # the source code of the file changed, we need to reload 74 checksum = pickle.load(f) 75 if self.checksum != checksum: 76 self.reset() 77 return 78 # if marshal_load fails then we need to reload 79 try: 80 self.code = marshal.load(f) 81 except (EOFError, ValueError, TypeError): 82 self.reset() 83 return 84 85 def write_bytecode(self, f: t.IO[bytes]) -> None: 86 """Dump the bytecode into the file or file like object passed.""" 87 if self.code is None: 88 raise TypeError("can't write empty bucket") 89 f.write(bc_magic) 90 pickle.dump(self.checksum, f, 2) 91 marshal.dump(self.code, f) 92 93 def bytecode_from_string(self, string: bytes) -> None: 94 """Load bytecode from bytes.""" 95 self.load_bytecode(BytesIO(string)) 96 97 def bytecode_to_string(self) -> bytes: 98 """Return the bytecode as bytes.""" 99 out = BytesIO() 100 self.write_bytecode(out) 101 return out.getvalue()
Buckets are used to store the bytecode for one template. It's created and initialized by the bytecode cache and passed to the loading functions.
The buckets get an internal checksum from the cache assigned and use this to automatically reject outdated cache material. Individual bytecode cache subclasses don't have to care about cache invalidation.
Bucket(environment: jinja2.environment.Environment, key: str, checksum: str)
def
reset(self) -> None:
62 def reset(self) -> None: 63 """Resets the bucket (unloads the bytecode).""" 64 self.code: t.Optional[CodeType] = None
Resets the bucket (unloads the bytecode).
def
load_bytecode(self, f: <class 'BinaryIO'>) -> None:
66 def load_bytecode(self, f: t.BinaryIO) -> None: 67 """Loads bytecode from a file or file like object.""" 68 # make sure the magic header is correct 69 magic = f.read(len(bc_magic)) 70 if magic != bc_magic: 71 self.reset() 72 return 73 # the source code of the file changed, we need to reload 74 checksum = pickle.load(f) 75 if self.checksum != checksum: 76 self.reset() 77 return 78 # if marshal_load fails then we need to reload 79 try: 80 self.code = marshal.load(f) 81 except (EOFError, ValueError, TypeError): 82 self.reset() 83 return
Loads bytecode from a file or file like object.
def
write_bytecode(self, f: IO[bytes]) -> None:
85 def write_bytecode(self, f: t.IO[bytes]) -> None: 86 """Dump the bytecode into the file or file like object passed.""" 87 if self.code is None: 88 raise TypeError("can't write empty bucket") 89 f.write(bc_magic) 90 pickle.dump(self.checksum, f, 2) 91 marshal.dump(self.code, f)
Dump the bytecode into the file or file like object passed.
class
BytecodeCache:
104class BytecodeCache: 105 """To implement your own bytecode cache you have to subclass this class 106 and override :meth:`load_bytecode` and :meth:`dump_bytecode`. Both of 107 these methods are passed a :class:`~jinja2.bccache.Bucket`. 108 109 A very basic bytecode cache that saves the bytecode on the file system:: 110 111 from os import path 112 113 class MyCache(BytecodeCache): 114 115 def __init__(self, directory): 116 self.directory = directory 117 118 def load_bytecode(self, bucket): 119 filename = path.join(self.directory, bucket.key) 120 if path.exists(filename): 121 with open(filename, 'rb') as f: 122 bucket.load_bytecode(f) 123 124 def dump_bytecode(self, bucket): 125 filename = path.join(self.directory, bucket.key) 126 with open(filename, 'wb') as f: 127 bucket.write_bytecode(f) 128 129 A more advanced version of a filesystem based bytecode cache is part of 130 Jinja. 131 """ 132 133 def load_bytecode(self, bucket: Bucket) -> None: 134 """Subclasses have to override this method to load bytecode into a 135 bucket. If they are not able to find code in the cache for the 136 bucket, it must not do anything. 137 """ 138 raise NotImplementedError() 139 140 def dump_bytecode(self, bucket: Bucket) -> None: 141 """Subclasses have to override this method to write the bytecode 142 from a bucket back to the cache. If it unable to do so it must not 143 fail silently but raise an exception. 144 """ 145 raise NotImplementedError() 146 147 def clear(self) -> None: 148 """Clears the cache. This method is not used by Jinja but should be 149 implemented to allow applications to clear the bytecode cache used 150 by a particular environment. 151 """ 152 153 def get_cache_key( 154 self, name: str, filename: t.Optional[t.Union[str]] = None 155 ) -> str: 156 """Returns the unique hash key for this template name.""" 157 hash = sha1(name.encode("utf-8")) 158 159 if filename is not None: 160 hash.update(f"|{filename}".encode()) 161 162 return hash.hexdigest() 163 164 def get_source_checksum(self, source: str) -> str: 165 """Returns a checksum for the source.""" 166 return sha1(source.encode("utf-8")).hexdigest() 167 168 def get_bucket( 169 self, 170 environment: "Environment", 171 name: str, 172 filename: t.Optional[str], 173 source: str, 174 ) -> Bucket: 175 """Return a cache bucket for the given template. All arguments are 176 mandatory but filename may be `None`. 177 """ 178 key = self.get_cache_key(name, filename) 179 checksum = self.get_source_checksum(source) 180 bucket = Bucket(environment, key, checksum) 181 self.load_bytecode(bucket) 182 return bucket 183 184 def set_bucket(self, bucket: Bucket) -> None: 185 """Put the bucket into the cache.""" 186 self.dump_bytecode(bucket)
To implement your own bytecode cache you have to subclass this class
and override load_bytecode() and dump_bytecode(). Both of
these methods are passed a ~Bucket.
A very basic bytecode cache that saves the bytecode on the file system::
from os import path
class MyCache(BytecodeCache):
def __init__(self, directory):
self.directory = directory
def load_bytecode(self, bucket):
filename = path.join(self.directory, bucket.key)
if path.exists(filename):
with open(filename, 'rb') as f:
bucket.load_bytecode(f)
def dump_bytecode(self, bucket):
filename = path.join(self.directory, bucket.key)
with open(filename, 'wb') as f:
bucket.write_bytecode(f)
A more advanced version of a filesystem based bytecode cache is part of Jinja.
133 def load_bytecode(self, bucket: Bucket) -> None: 134 """Subclasses have to override this method to load bytecode into a 135 bucket. If they are not able to find code in the cache for the 136 bucket, it must not do anything. 137 """ 138 raise NotImplementedError()
Subclasses have to override this method to load bytecode into a bucket. If they are not able to find code in the cache for the bucket, it must not do anything.
140 def dump_bytecode(self, bucket: Bucket) -> None: 141 """Subclasses have to override this method to write the bytecode 142 from a bucket back to the cache. If it unable to do so it must not 143 fail silently but raise an exception. 144 """ 145 raise NotImplementedError()
Subclasses have to override this method to write the bytecode from a bucket back to the cache. If it unable to do so it must not fail silently but raise an exception.
def
clear(self) -> None:
147 def clear(self) -> None: 148 """Clears the cache. This method is not used by Jinja but should be 149 implemented to allow applications to clear the bytecode cache used 150 by a particular environment. 151 """
Clears the cache. This method is not used by Jinja but should be implemented to allow applications to clear the bytecode cache used by a particular environment.
def
get_cache_key(self, name: str, filename: Optional[str] = None) -> str:
153 def get_cache_key( 154 self, name: str, filename: t.Optional[t.Union[str]] = None 155 ) -> str: 156 """Returns the unique hash key for this template name.""" 157 hash = sha1(name.encode("utf-8")) 158 159 if filename is not None: 160 hash.update(f"|{filename}".encode()) 161 162 return hash.hexdigest()
Returns the unique hash key for this template name.
def
get_source_checksum(self, source: str) -> str:
164 def get_source_checksum(self, source: str) -> str: 165 """Returns a checksum for the source.""" 166 return sha1(source.encode("utf-8")).hexdigest()
Returns a checksum for the source.
def
get_bucket( self, environment: jinja2.environment.Environment, name: str, filename: Optional[str], source: str) -> Bucket:
168 def get_bucket( 169 self, 170 environment: "Environment", 171 name: str, 172 filename: t.Optional[str], 173 source: str, 174 ) -> Bucket: 175 """Return a cache bucket for the given template. All arguments are 176 mandatory but filename may be `None`. 177 """ 178 key = self.get_cache_key(name, filename) 179 checksum = self.get_source_checksum(source) 180 bucket = Bucket(environment, key, checksum) 181 self.load_bytecode(bucket) 182 return bucket
Return a cache bucket for the given template. All arguments are
mandatory but filename may be None.
189class FileSystemBytecodeCache(BytecodeCache): 190 """A bytecode cache that stores bytecode on the filesystem. It accepts 191 two arguments: The directory where the cache items are stored and a 192 pattern string that is used to build the filename. 193 194 If no directory is specified a default cache directory is selected. On 195 Windows the user's temp directory is used, on UNIX systems a directory 196 is created for the user in the system temp directory. 197 198 The pattern can be used to have multiple separate caches operate on the 199 same directory. The default pattern is ``'__jinja2_%s.cache'``. ``%s`` 200 is replaced with the cache key. 201 202 >>> bcc = FileSystemBytecodeCache('/tmp/jinja_cache', '%s.cache') 203 204 This bytecode cache supports clearing of the cache using the clear method. 205 """ 206 207 def __init__( 208 self, directory: t.Optional[str] = None, pattern: str = "__jinja2_%s.cache" 209 ) -> None: 210 if directory is None: 211 directory = self._get_default_cache_dir() 212 self.directory = directory 213 self.pattern = pattern 214 215 def _get_default_cache_dir(self) -> str: 216 def _unsafe_dir() -> "te.NoReturn": 217 raise RuntimeError( 218 "Cannot determine safe temp directory. You " 219 "need to explicitly provide one." 220 ) 221 222 tmpdir = tempfile.gettempdir() 223 224 # On windows the temporary directory is used specific unless 225 # explicitly forced otherwise. We can just use that. 226 if os.name == "nt": 227 return tmpdir 228 if not hasattr(os, "getuid"): 229 _unsafe_dir() 230 231 dirname = f"_jinja2-cache-{os.getuid()}" 232 actual_dir = os.path.join(tmpdir, dirname) 233 234 try: 235 os.mkdir(actual_dir, stat.S_IRWXU) 236 except OSError as e: 237 if e.errno != errno.EEXIST: 238 raise 239 try: 240 os.chmod(actual_dir, stat.S_IRWXU) 241 actual_dir_stat = os.lstat(actual_dir) 242 if ( 243 actual_dir_stat.st_uid != os.getuid() 244 or not stat.S_ISDIR(actual_dir_stat.st_mode) 245 or stat.S_IMODE(actual_dir_stat.st_mode) != stat.S_IRWXU 246 ): 247 _unsafe_dir() 248 except OSError as e: 249 if e.errno != errno.EEXIST: 250 raise 251 252 actual_dir_stat = os.lstat(actual_dir) 253 if ( 254 actual_dir_stat.st_uid != os.getuid() 255 or not stat.S_ISDIR(actual_dir_stat.st_mode) 256 or stat.S_IMODE(actual_dir_stat.st_mode) != stat.S_IRWXU 257 ): 258 _unsafe_dir() 259 260 return actual_dir 261 262 def _get_cache_filename(self, bucket: Bucket) -> str: 263 return os.path.join(self.directory, self.pattern % (bucket.key,)) 264 265 def load_bytecode(self, bucket: Bucket) -> None: 266 filename = self._get_cache_filename(bucket) 267 268 # Don't test for existence before opening the file, since the 269 # file could disappear after the test before the open. 270 try: 271 f = open(filename, "rb") 272 except (FileNotFoundError, IsADirectoryError, PermissionError): 273 # PermissionError can occur on Windows when an operation is 274 # in progress, such as calling clear(). 275 return 276 277 with f: 278 bucket.load_bytecode(f) 279 280 def dump_bytecode(self, bucket: Bucket) -> None: 281 # Write to a temporary file, then rename to the real name after 282 # writing. This avoids another process reading the file before 283 # it is fully written. 284 name = self._get_cache_filename(bucket) 285 f = tempfile.NamedTemporaryFile( 286 mode="wb", 287 dir=os.path.dirname(name), 288 prefix=os.path.basename(name), 289 suffix=".tmp", 290 delete=False, 291 ) 292 293 def remove_silent() -> None: 294 try: 295 os.remove(f.name) 296 except OSError: 297 # Another process may have called clear(). On Windows, 298 # another program may be holding the file open. 299 pass 300 301 try: 302 with f: 303 bucket.write_bytecode(f) 304 except BaseException: 305 remove_silent() 306 raise 307 308 try: 309 os.replace(f.name, name) 310 except OSError: 311 # Another process may have called clear(). On Windows, 312 # another program may be holding the file open. 313 remove_silent() 314 except BaseException: 315 remove_silent() 316 raise 317 318 def clear(self) -> None: 319 # imported lazily here because google app-engine doesn't support 320 # write access on the file system and the function does not exist 321 # normally. 322 from os import remove 323 324 files = fnmatch.filter(os.listdir(self.directory), self.pattern % ("*",)) 325 for filename in files: 326 try: 327 remove(os.path.join(self.directory, filename)) 328 except OSError: 329 pass
A bytecode cache that stores bytecode on the filesystem. It accepts two arguments: The directory where the cache items are stored and a pattern string that is used to build the filename.
If no directory is specified a default cache directory is selected. On Windows the user's temp directory is used, on UNIX systems a directory is created for the user in the system temp directory.
The pattern can be used to have multiple separate caches operate on the
same directory. The default pattern is '__jinja2_%s.cache'. %s
is replaced with the cache key.
>>> bcc = FileSystemBytecodeCache('/tmp/jinja_cache', '%s.cache')
This bytecode cache supports clearing of the cache using the clear method.
265 def load_bytecode(self, bucket: Bucket) -> None: 266 filename = self._get_cache_filename(bucket) 267 268 # Don't test for existence before opening the file, since the 269 # file could disappear after the test before the open. 270 try: 271 f = open(filename, "rb") 272 except (FileNotFoundError, IsADirectoryError, PermissionError): 273 # PermissionError can occur on Windows when an operation is 274 # in progress, such as calling clear(). 275 return 276 277 with f: 278 bucket.load_bytecode(f)
Subclasses have to override this method to load bytecode into a bucket. If they are not able to find code in the cache for the bucket, it must not do anything.
280 def dump_bytecode(self, bucket: Bucket) -> None: 281 # Write to a temporary file, then rename to the real name after 282 # writing. This avoids another process reading the file before 283 # it is fully written. 284 name = self._get_cache_filename(bucket) 285 f = tempfile.NamedTemporaryFile( 286 mode="wb", 287 dir=os.path.dirname(name), 288 prefix=os.path.basename(name), 289 suffix=".tmp", 290 delete=False, 291 ) 292 293 def remove_silent() -> None: 294 try: 295 os.remove(f.name) 296 except OSError: 297 # Another process may have called clear(). On Windows, 298 # another program may be holding the file open. 299 pass 300 301 try: 302 with f: 303 bucket.write_bytecode(f) 304 except BaseException: 305 remove_silent() 306 raise 307 308 try: 309 os.replace(f.name, name) 310 except OSError: 311 # Another process may have called clear(). On Windows, 312 # another program may be holding the file open. 313 remove_silent() 314 except BaseException: 315 remove_silent() 316 raise
Subclasses have to override this method to write the bytecode from a bucket back to the cache. If it unable to do so it must not fail silently but raise an exception.
def
clear(self) -> None:
318 def clear(self) -> None: 319 # imported lazily here because google app-engine doesn't support 320 # write access on the file system and the function does not exist 321 # normally. 322 from os import remove 323 324 files = fnmatch.filter(os.listdir(self.directory), self.pattern % ("*",)) 325 for filename in files: 326 try: 327 remove(os.path.join(self.directory, filename)) 328 except OSError: 329 pass
Clears the cache. This method is not used by Jinja but should be implemented to allow applications to clear the bytecode cache used by a particular environment.
Inherited Members
332class MemcachedBytecodeCache(BytecodeCache): 333 """This class implements a bytecode cache that uses a memcache cache for 334 storing the information. It does not enforce a specific memcache library 335 (tummy's memcache or cmemcache) but will accept any class that provides 336 the minimal interface required. 337 338 Libraries compatible with this class: 339 340 - `cachelib <https://github.com/pallets/cachelib>`_ 341 - `python-memcached <https://pypi.org/project/python-memcached/>`_ 342 343 (Unfortunately the django cache interface is not compatible because it 344 does not support storing binary data, only text. You can however pass 345 the underlying cache client to the bytecode cache which is available 346 as `django.core.cache.cache._client`.) 347 348 The minimal interface for the client passed to the constructor is this: 349 350 .. class:: MinimalClientInterface 351 352 .. method:: set(key, value[, timeout]) 353 354 Stores the bytecode in the cache. `value` is a string and 355 `timeout` the timeout of the key. If timeout is not provided 356 a default timeout or no timeout should be assumed, if it's 357 provided it's an integer with the number of seconds the cache 358 item should exist. 359 360 .. method:: get(key) 361 362 Returns the value for the cache key. If the item does not 363 exist in the cache the return value must be `None`. 364 365 The other arguments to the constructor are the prefix for all keys that 366 is added before the actual cache key and the timeout for the bytecode in 367 the cache system. We recommend a high (or no) timeout. 368 369 This bytecode cache does not support clearing of used items in the cache. 370 The clear method is a no-operation function. 371 372 .. versionadded:: 2.7 373 Added support for ignoring memcache errors through the 374 `ignore_memcache_errors` parameter. 375 """ 376 377 def __init__( 378 self, 379 client: "_MemcachedClient", 380 prefix: str = "jinja2/bytecode/", 381 timeout: t.Optional[int] = None, 382 ignore_memcache_errors: bool = True, 383 ): 384 self.client = client 385 self.prefix = prefix 386 self.timeout = timeout 387 self.ignore_memcache_errors = ignore_memcache_errors 388 389 def load_bytecode(self, bucket: Bucket) -> None: 390 try: 391 code = self.client.get(self.prefix + bucket.key) 392 except Exception: 393 if not self.ignore_memcache_errors: 394 raise 395 else: 396 bucket.bytecode_from_string(code) 397 398 def dump_bytecode(self, bucket: Bucket) -> None: 399 key = self.prefix + bucket.key 400 value = bucket.bytecode_to_string() 401 402 try: 403 if self.timeout is not None: 404 self.client.set(key, value, self.timeout) 405 else: 406 self.client.set(key, value) 407 except Exception: 408 if not self.ignore_memcache_errors: 409 raise
This class implements a bytecode cache that uses a memcache cache for storing the information. It does not enforce a specific memcache library (tummy's memcache or cmemcache) but will accept any class that provides the minimal interface required.
Libraries compatible with this class:
(Unfortunately the django cache interface is not compatible because it
does not support storing binary data, only text. You can however pass
the underlying cache client to the bytecode cache which is available
as django.core.cache.cache._client.)
The minimal interface for the client passed to the constructor is this:
.. class:: MinimalClientInterface
.. method:: set(key, value[, timeout])
Stores the bytecode in the cache. `value` is a string and
`timeout` the timeout of the key. If timeout is not provided
a default timeout or no timeout should be assumed, if it's
provided it's an integer with the number of seconds the cache
item should exist.
.. method:: get(key)
Returns the value for the cache key. If the item does not
exist in the cache the return value must be `None`.
The other arguments to the constructor are the prefix for all keys that is added before the actual cache key and the timeout for the bytecode in the cache system. We recommend a high (or no) timeout.
This bytecode cache does not support clearing of used items in the cache. The clear method is a no-operation function.
New in version 2.7:
Added support for ignoring memcache errors through the
ignore_memcache_errors parameter.
MemcachedBytecodeCache( client: jinja2.bccache._MemcachedClient, prefix: str = 'jinja2/bytecode/', timeout: Optional[int] = None, ignore_memcache_errors: bool = True)
377 def __init__( 378 self, 379 client: "_MemcachedClient", 380 prefix: str = "jinja2/bytecode/", 381 timeout: t.Optional[int] = None, 382 ignore_memcache_errors: bool = True, 383 ): 384 self.client = client 385 self.prefix = prefix 386 self.timeout = timeout 387 self.ignore_memcache_errors = ignore_memcache_errors
389 def load_bytecode(self, bucket: Bucket) -> None: 390 try: 391 code = self.client.get(self.prefix + bucket.key) 392 except Exception: 393 if not self.ignore_memcache_errors: 394 raise 395 else: 396 bucket.bytecode_from_string(code)
Subclasses have to override this method to load bytecode into a bucket. If they are not able to find code in the cache for the bucket, it must not do anything.
398 def dump_bytecode(self, bucket: Bucket) -> None: 399 key = self.prefix + bucket.key 400 value = bucket.bytecode_to_string() 401 402 try: 403 if self.timeout is not None: 404 self.client.set(key, value, self.timeout) 405 else: 406 self.client.set(key, value) 407 except Exception: 408 if not self.ignore_memcache_errors: 409 raise
Subclasses have to override this method to write the bytecode from a bucket back to the cache. If it unable to do so it must not fail silently but raise an exception.