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_new_invalid_cache_raises(self):
with pytest.raises(InvalidCacheType) as e:
Cache(object)
assert str(e.value) == "Invalid cache type, you can only use {}".format(
list(AIOCACHE_CACHES.keys())
)
import asyncio
import argparse
import logging
import uuid
from aiocache import Cache
from aiohttp import web
logging.getLogger("aiohttp.access").propagate = False
AIOCACHE_BACKENDS = {
"memory": Cache(Cache.MEMORY),
"redis": Cache(Cache.REDIS),
"memcached": Cache(Cache.MEMCACHED),
}
class CacheManager:
def __init__(self, backend):
self.cache = AIOCACHE_BACKENDS.get(backend)
async def get(self, key):
return await self.cache.get(key, timeout=0.1)
async def set(self, key, value):
return await self.cache.set(key, value, timeout=0.1)
async def handler_get(req):
try:
import asyncio
import logging
from aiocache import Cache
from aiocache.lock import RedLock
logger = logging.getLogger(__name__)
cache = Cache(Cache.REDIS, endpoint='127.0.0.1', port=6379, namespace='main')
async def expensive_function():
logger.warning('Expensive is being executed...')
await asyncio.sleep(1)
return 'result'
async def my_view():
async with RedLock(cache, 'key', lease=2): # Wait at most 2 seconds
result = await cache.get('key')
if result is not None:
logger.info('Found the value in the cache hurray!')
return result
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)