Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"default": {
"cache": "aiocache.RedisCache",
"endpoint": "127.0.0.10",
"port": 6378,
"serializer": {"class": "aiocache.serializers.PickleSerializer"},
"plugins": [
{"class": "aiocache.plugins.HitMissRatioPlugin"},
{"class": "aiocache.plugins.TimingPlugin"},
],
},
"alt": {"cache": "aiocache.SimpleMemoryCache"},
}
)
default = caches.create(**caches.get_alias_config("default"))
alt = caches.create(**caches.get_alias_config("alt"))
assert isinstance(default, RedisCache)
assert default.endpoint == "127.0.0.10"
assert default.port == 6378
assert isinstance(default.serializer, PickleSerializer)
assert len(default.plugins) == 2
assert isinstance(alt, SimpleMemoryCache)
{
"default": {
"cache": "aiocache.RedisCache",
"endpoint": "127.0.0.10",
"port": 6378,
"serializer": {"class": "aiocache.serializers.PickleSerializer"},
"plugins": [
{"class": "aiocache.plugins.HitMissRatioPlugin"},
{"class": "aiocache.plugins.TimingPlugin"},
],
},
"alt": {"cache": "aiocache.SimpleMemoryCache"},
}
)
default = caches.create(**caches.get_alias_config("default"))
alt = caches.create(**caches.get_alias_config("alt"))
assert isinstance(default, RedisCache)
assert default.endpoint == "127.0.0.10"
assert default.port == 6378
assert isinstance(default.serializer, PickleSerializer)
assert len(default.plugins) == 2
assert isinstance(alt, SimpleMemoryCache)
def test_get_alias_config(self):
assert caches.get_alias_config("default") == {
"cache": "aiocache.SimpleMemoryCache",
"serializer": {"class": "aiocache.serializers.NullSerializer"},
}
def get_alias(cls, alias):
logger.warning("Deprecated, please use 'aiocache.caches.get_config[alias]'")
return caches.get_alias_config(alias)
async def alt_cache():
# This generates a new instance every time! You can also use `caches.create('alt')`
# or even `caches.create('alt', namespace="test", etc...)` to override extra args
cache = caches.create(**caches.get_alias_config('redis_alt'))
await cache.set("key", "value")
assert await cache.get("key") == "value"
assert isinstance(cache, Cache.REDIS)
assert isinstance(cache.serializer, PickleSerializer)
assert len(cache.plugins) == 2
assert cache.endpoint == "127.0.0.1"
assert cache.timeout == 1
assert cache.port == 6379
await cache.close()
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)