Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_cache(namespace=None):
if namespace is None:
return aiocache.caches.get('default')
try:
return aiocache.caches.get(namespace)
except KeyError:
pass
_cache_config = aiocache.caches.get_alias_config('default')
base_namespace = _cache_config['namespace']
if not base_namespace:
actual_namespace = namespace
else:
actual_namespace = '_'.join((base_namespace, namespace))
_cache_config['namespace'] = actual_namespace
aiocache.caches.add(namespace, _cache_config)
return aiocache.caches.get(namespace)
return aiocache.caches.get('default')
try:
return aiocache.caches.get(namespace)
except KeyError:
pass
_cache_config = aiocache.caches.get_alias_config('default')
base_namespace = _cache_config['namespace']
if not base_namespace:
actual_namespace = namespace
else:
actual_namespace = '_'.join((base_namespace, namespace))
_cache_config['namespace'] = actual_namespace
aiocache.caches.add(namespace, _cache_config)
return aiocache.caches.get(namespace)
def test_alias():
loop = asyncio.get_event_loop()
loop.run_until_complete(default_cache())
loop.run_until_complete(alt_cache())
cache = Cache(Cache.REDIS)
loop.run_until_complete(cache.delete("key"))
loop.run_until_complete(cache.close())
loop.run_until_complete(caches.get('default').close())
async def default_cache():
cache = caches.get('default') # This always returns the same instance
await cache.set("key", "value")
assert await cache.get("key") == "value"
assert isinstance(cache, Cache.MEMORY)
assert isinstance(cache.serializer, StringSerializer)
def __call__(self, f):
if self.alias:
self.cache = caches.get(self.alias)
else:
self.cache = _get_cache(
cache=self._cache,
serializer=self._serializer,
plugins=self._plugins,
**self._kwargs
)
@functools.wraps(f)
async def wrapper(*args, **kwargs):
return await self.decorator(f, *args, **kwargs)
wrapper.cache = self.cache
return wrapper
def init_cache(sanic, loop):
caches.set_config(config)
if namespace is None:
return aiocache.caches.get('default')
try:
return aiocache.caches.get(namespace)
except KeyError:
pass
_cache_config = aiocache.caches.get_alias_config('default')
base_namespace = _cache_config['namespace']
if not base_namespace:
actual_namespace = namespace
else:
actual_namespace = '_'.join((base_namespace, namespace))
_cache_config['namespace'] = actual_namespace
aiocache.caches.add(namespace, _cache_config)
return aiocache.caches.get(namespace)
async def get_cache_value():
# This lazy loads a singleton so it will return the same instance every
# time. If you want to create a new instance, you can use
# `caches.create("default")`
cache = caches.get("default")
return await cache.get("my_custom_key")