Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.backend.set(key, value, expire)
def delete_value(self, key):
try:
self.backend.delete(key)
except KeyError:
pass
def touch_value(self, key, expire):
try:
self.backend.touch(key)
except KeyError:
pass
class ExpirableDictStorage(fbase.CommonMixinStorage, fbase.StorageMixin):
in_memory_storage = True
now = time.time
def get_value(self, key):
_now = self.now()
try:
expired_time, value = self.backend[key]
except KeyError:
raise fbase.NotFound
if expired_time is not None and expired_time < _now:
raise fbase.NotFound
return value
def set_value(self, key, value, expire):
_now = self.now()
self.set_many_values(
keys, [self.rope.encode(v) for v in values], expire)
def delete_many(self, keys):
self.delete_many_values(keys)
def has_many(self, keys):
return self.has_many_values(keys)
def touch_many(self, keys, expire=Ellipsis):
if expire is Ellipsis:
expire = self.rope.config.expire_default
self.touch_many_values(keys, expire)
class LruStorage(fbase.CommonMixinStorage, fbase.StorageMixin):
in_memory_storage = True
def get_value(self, key):
value = self.backend.get(key)
if value is lru_mod.SENTINEL:
raise fbase.NotFound
return value
def has_value(self, key):
return self.backend.has(key)
def set_value(self, key, value, expire):
self.backend.set(key, value, expire)
def delete_value(self, key):
from django.middleware.cache import CacheMiddleware
from .func import base as fbase
from .func.sync import CacheUserInterface
__all__ = ('cache', 'cache_page')
def promote_backend(backend):
"""Get string name to Django cache backend."""
if isinstance(backend, (str, bytes)):
backend = django_cache.caches[backend]
return backend
class LowLevelCacheStorage(fbase.CommonMixinStorage, fbase.StorageMixin):
"""Storage implementation for :data:`django.core.cache.caches`."""
def get_value(self, key):
value = self.backend.get(key)
if value is None:
raise fbase.NotFound
return value
def set_value(self, key, value, expire):
self.backend.set(key, value, timeout=expire)
def delete_value(self, key):
self.backend.delete(key)
def transform_cache_page_args(wire, rules, args, kwargs):
def convert_storage(storage_class):
storage_bases = (fbase.CommonMixinStorage, BulkStorageMixin)
async_storage_class = type(
'Async' + storage_class.__name__, (storage_class,), {})
count = 0
for storage_base in storage_bases:
if issubclass(storage_class, storage_base):
count += 1
for name in storage_base.__dict__.keys():
async_attr = asyncio.coroutine(getattr(storage_class, name))
setattr(async_storage_class, name, async_attr)
if count == 0:
raise TypeError(
"'storage_class' is not subclassing any known storage base")
return async_storage_class
def delete_value(self, key):
self.backend.hdel(self.hash_key, key)
def has_value(self, key):
return self.backend.hexists(self.hash_key, key)
def get_many_values(self, keys):
values = self.backend.hmget(self.hash_key, keys)
return [v if v is not None else fbase.NotFound for v in values]
def set_many_values(self, keys, values, expire):
self.backend.hmset(self.hash_key, {k: v for k, v in zip(keys, values)})
class DiskCacheStorage(fbase.CommonMixinStorage, fbase.StorageMixin):
def get_value(self, key):
value = self.backend.get(key)
if value is None:
raise fbase.NotFound
return value
def set_value(self, key, value, expire):
self.backend.set(key, value, expire)
def delete_value(self, key):
self.backend.delete(key)
def lru(
lru=None, key_prefix=None, expire=None, coder=None,
def touch_value(self, key, expire):
self.backend.touch(key, expire)
def get_many_values(self, keys):
values = self.backend.get_multi(keys)
return [values.get(k, fbase.NotFound) for k in keys]
def set_many_values(self, keys, values, expire):
self.backend.set_multi({k: v for k, v in zip(keys, values)}, expire)
def delete_many_values(self, keys):
return self.backend.delete_multi(keys)
class RedisStorage(
fbase.CommonMixinStorage, fbase.StorageMixin, BulkStorageMixin):
def get_value(self, key):
value = self.backend.get(key)
if value is None:
raise fbase.NotFound
return value
def set_value(self, key, value, expire):
self.backend.set(key, value, expire)
def delete_value(self, key):
self.backend.delete(key)
def has_value(self, key):
return bool(self.backend.exists(key))
return key in self.backend
def touch_value(self, key, expire):
_now = self.now()
try:
expired_time, value = self.backend[key]
except KeyError:
return
if expire is None:
expired_time = None
else:
expired_time = _now + expire
self.backend[key] = expired_time, value
class PersistentDictStorage(fbase.CommonMixinStorage, fbase.StorageMixin):
in_memory_storage = True
def get_value(self, key):
try:
value = self.backend[key]
except KeyError:
raise fbase.NotFound
return value
def set_value(self, key, value, expire):
self.backend[key] = value
def delete_value(self, key):
try:
del self.backend[key]
class ShelveStorage(PersistentDictStorage):
def set_value(self, key, value, expire):
super(ShelveStorage, self).set_value(key, value, expire)
self.backend.sync()
def delete_value(self, key):
super(ShelveStorage, self).delete_value(key)
self.backend.sync()
def __del__(self):
self.backend.close()
class MemcacheStorage(
fbase.CommonMixinStorage, fbase.StorageMixin, BulkStorageMixin):
def get_value(self, key):
value = self.backend.get(key)
if value is None:
raise fbase.NotFound
return value
def set_value(self, key, value, expire):
self.backend.set(key, value, expire)
def delete_value(self, key):
self.backend.delete(key)
def touch_value(self, key, expire):
self.backend.touch(key, expire)
def arcus(
client, key_prefix=None, expire=0, coder=None,
default_action='get_or_update',
user_interface=CacheUserInterface,
**kwargs): # pragma: no cover
"""Arcus support. deprecated."""
class Storage(fbase.CommonMixinStorage, fbase.StorageMixin):
def get_value(self, key):
value = self.backend.get(key).get_result()
if value is None:
raise fbase.NotFound
return value
def set_value(self, key, value, expire):
self.backend.set(key, value, expire)
def delete_value(self, key):
self.backend.delete(key)
def touch_value(self, key, expire):
self.backend.touch(key, expire)
rule = re.compile(r'[!-~]+')