Source code for Aeros.patches.flask_caching.Cache

from flask_caching import Cache as OriginalCache
from flask_caching import *

[docs]class Cache(OriginalCache): def __init__(self, timeout: int = 60 * 60, threshold: int = 100, *args, **kwargs): OriginalCache.__init__(self, config={}, *args, **kwargs) self.config["CACHE_TYPE"] = "null" self.config["CACHE_THRESHOLD"] = threshold self.config["CACHE_DEFAULT_TIMEOUT"] = timeout
[docs] def cached( self, timeout=None, key_prefix="view/%s", unless=None, forced_update=None, response_filter=None, query_string=False, hash_method=hashlib.md5, cache_none=False, ): """Decorator. Use this to cache a function. By default the cache key is `view/request.path`. You are able to use this decorator with any function by changing the `key_prefix`. If the token `%s` is located within the `key_prefix` then it will replace that with `request.path` Example:: # An example view function @cache.cached(timeout=50) def big_foo(): return big_bar_calc() # An example misc function to cache. @cache.cached(key_prefix='MyCachedList') def get_list(): return [random.randrange(0, 1) for i in range(50000)] my_list = get_list() .. note:: You MUST have a request context to actually called any functions that are cached. .. versionadded:: 0.4 The returned decorated function now has three function attributes assigned to it. These attributes are readable/writable. **uncached** The original undecorated function **cache_timeout** The cache timeout value for this function. For a custom value to take affect, this must be set before the function is called. **make_cache_key** A function used in generating the cache_key used. :param timeout: Default None. If set to an integer, will cache for that amount of time. Unit of time is in seconds. :param key_prefix: Default 'view/%(request.path)s'. Beginning key to . use for the cache key. `request.path` will be the actual request path, or in cases where the `make_cache_key`-function is called from other views it will be the expected URL for the view as generated by Flask's `url_for()`. .. versionadded:: 0.3.4 Can optionally be a callable which takes no arguments but returns a string that will be used as the cache_key. :param unless: Default None. Cache will *always* execute the caching facilities unless this callable is true. This will bypass the caching entirely. :param forced_update: Default None. If this callable is true, cache value will be updated regardless cache is expired or not. Useful for background renewal of cached functions. :param response_filter: Default None. If not None, the callable is invoked after the cached funtion evaluation, and is given one arguement, the response content. If the callable returns False, the content will not be cached. Useful to prevent caching of code 500 responses. :param query_string: Default False. When True, the cache key used will be the result of hashing the ordered query string parameters. This avoids creating different caches for the same query just because the parameters were passed in a different order. See _make_cache_key_query_string() for more details. :param hash_method: Default hashlib.md5. The hash method used to generate the keys for cached results. :param cache_none: Default False. If set to True, add a key exists check when cache.get returns None. This will likely lead to wrongly returned None values in concurrent situations and is not recommended to use. """ def decorator(f): @functools.wraps(f) async def decorated_function(*args, **kwargs): #: Bypass the cache entirely. if self._bypass_cache(unless, f, *args, **kwargs): return await f(*args, **kwargs) try: cache_key = "{}{}{}".format(f.__name__, args, kwargs) if ( callable(forced_update) and ( forced_update(*args, **kwargs) if wants_args(forced_update) else forced_update() ) is True ): rv = None found = False else: rv = self.cache.get(cache_key) found = True # If the value returned by cache.get() is None, it # might be because the key is not found in the cache # or because the cached value is actually None if rv is None: # If we're sure we don't need to cache None values # (cache_none=False), don't bother checking for # key existence, as it can lead to false positives # if a concurrent call already cached the # key between steps. This would cause us to # return None when we shouldn't if not cache_none: found = False else: found = self.cache.has(cache_key) except Exception: if raise logger.exception("Exception possibly due to cache backend.") return await f(*args, **kwargs) if not found: rv = await f(*args, **kwargs) if response_filter is None or response_filter(rv): try: self.cache.set( cache_key, rv, timeout=decorated_function.cache_timeout, ) except Exception: if raise logger.exception( "Exception possibly due to cache backend." ) return rv def make_cache_key(*args, **kwargs): # Convert non-keyword arguments (which is the way # `make_cache_key` expects them) to keyword arguments # (the way `url_for` expects them) argspec_args = inspect.getfullargspec(f).args for arg_name, arg in zip(argspec_args, args): kwargs[arg_name] = arg return _make_cache_key(args, kwargs, use_request=False) def _make_cache_key_query_string(): """Create consistent keys for query string arguments. Produces the same cache key regardless of argument order, e.g., both `?limit=10&offset=20` and `?offset=20&limit=10` will always produce the same exact cache key. """ # Create a tuple of (key, value) pairs, where the key is the # argument name and the value is its respective value. Order # this tuple by key. Doing this ensures the cache key created # is always the same for query string args whose keys/values # are the same, regardless of the order in which they are # provided. args_as_sorted_tuple = tuple( sorted((pair for pair in request.args.items(multi=True))) ) # ... now hash the sorted (key, value) tuple so it can be # used as a key for cache. Turn them into bytes so that the # hash function will accept them args_as_bytes = str(args_as_sorted_tuple).encode() hashed_args = str(hash_method(args_as_bytes).hexdigest()) cache_key = request.path + hashed_args return cache_key def _make_cache_key(args, kwargs, use_request): if callable(key_prefix): cache_key = key_prefix() elif "%s" in key_prefix: if use_request: cache_key = key_prefix % request.path else: cache_key = key_prefix % url_for(f.__name__, **kwargs) else: cache_key = key_prefix return cache_key decorated_function.uncached = f decorated_function.cache_timeout = timeout decorated_function.make_cache_key = make_cache_key return decorated_function return decorator