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