Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_decorator(self):
cache = self.cache(2)
wrapper = cachetools.cached(cache)(self.func)
self.assertEqual(len(cache), 0)
self.assertEqual(wrapper.__wrapped__, self.func)
self.assertEqual(wrapper(0), 0)
self.assertEqual(len(cache), 1)
self.assertIn(cachetools.keys.hashkey(0), cache)
self.assertNotIn(cachetools.keys.hashkey(1), cache)
self.assertNotIn(cachetools.keys.hashkey(1.0), cache)
self.assertEqual(wrapper(1), 1)
self.assertEqual(len(cache), 2)
self.assertIn(cachetools.keys.hashkey(0), cache)
self.assertIn(cachetools.keys.hashkey(1), cache)
self.assertIn(cachetools.keys.hashkey(1.0), cache)
self.assertEqual(wrapper(1), 1)
self.assertEqual(len(cache), 2)
self.assertEqual(wrapper(1.0), 1)
self.assertEqual(len(cache), 2)
self.assertEqual(wrapper(1.0), 1)
def _cache_ops_length(graph):
"""
Cachetools hashkey function that hashes graph properties based on len(graph.ops)
Arguments:
graph (ComputationalGraph): A computational graph instance
Returns:
A hashkey using cachetools.keys.hashkey
"""
return keys.hashkey(graph, len(graph.ops))
@cachetools.cached(cachetools.TTLCache(100, 60*30), key=lambda wg, url: cachetools.keys.hashkey(url))
def get_json(wg, url):
def cache_workflow_spec_by_execution_id(wf_ex_id, wf_spec):
with _WF_EX_CACHE_LOCK:
_WF_EX_CACHE[cachetools.keys.hashkey(wf_ex_id)] = wf_spec
key=lambda self, name: keys.hashkey(
self.work.collection._name
if self.work._collection is not None else None,
self.work.model_name,
name)
)
if use_args:
if not isinstance(use_args, (six.string_types, tuple)):
raise ValueError(
"'use_args' must be either a tuple or a string,"
" actual type: %s" % type(use_args)
)
use_args_tup = (
use_args if isinstance(use_args, tuple) else (use_args,)
)
for arg_name in arg_dict.keys():
if arg_name not in tuple(use_args_tup):
arg_dict.pop(arg_name, None)
return cachetools_keys.hashkey(**arg_dict)
def wrapper(*args, **kwargs):
if cache is not None:
args = [tuple(x) if isinstance(x, list) else x for x in args] # tuples are hashable
k = cachetools.keys.hashkey(GsSession.current, *args, **kwargs)
with metalock:
invocation_lock = invocation_locks.setdefault(f'{fn.__name__}:{k}', threading.Lock())
with invocation_lock:
with _fn_cache_lock:
result = cache.get(k)
if result:
_logger.debug('%s cache hit: %s, %s', fn.__name__, str(args), str(kwargs))
return result
result = fn(*args, **kwargs)
with _fn_cache_lock:
cache[k] = result
else:
result = fn(*args, **kwargs)
return result