Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return _indexes[name]
except KeyError:
parts = name.split('/')
directory = op.join(self._directory, 'index', *parts)
temp = Index(directory)
_indexes[name] = temp
return temp
############################################################################
# BEGIN Python 2/3 Shims
############################################################################
if sys.hexversion < 0x03000000:
import types
memoize_func = Cache.__dict__['memoize'] # pylint: disable=invalid-name
FanoutCache.memoize = types.MethodType(memoize_func, None, FanoutCache)
else:
FanoutCache.memoize = Cache.memoize
directory = op.join(self._directory, 'index', *parts)
temp = Index(directory)
_indexes[name] = temp
return temp
############################################################################
# BEGIN Python 2/3 Shims
############################################################################
if sys.hexversion < 0x03000000:
import types
memoize_func = Cache.__dict__['memoize'] # pylint: disable=invalid-name
FanoutCache.memoize = types.MethodType(memoize_func, None, FanoutCache)
else:
FanoutCache.memoize = Cache.memoize
def __init__(self, cache=None, expire=None):
if isinstance(cache, Cache):
pass
elif cache is None:
cache = Cache(tempfile.mkdtemp())
else:
cache = Cache(cache)
self._cache = cache
self._expire = expire
def __init__(self, iterable=(), directory=None):
"""Initialize deque instance.
If directory is None then temporary directory created. The directory
will *not* be automatically removed.
:param iterable: iterable of items to append to deque
:param directory: deque directory (default None)
"""
self._cache = Cache(directory, eviction_policy='none')
with self.transact():
self.extend(iterable)
:param disk: `Disk` instance for serialization
:param settings: any of `DEFAULT_SETTINGS`
"""
if directory is None:
directory = tempfile.mkdtemp(prefix='diskcache-')
directory = op.expanduser(directory)
directory = op.expandvars(directory)
default_size_limit = DEFAULT_SETTINGS['size_limit']
size_limit = settings.pop('size_limit', default_size_limit) / shards
self._count = shards
self._directory = directory
self._shards = tuple(
Cache(
directory=op.join(directory, '%03d' % num),
timeout=timeout,
disk=disk,
size_limit=size_limit,
**settings
)
for num in range(shards)
)
self._hash = self._shards[0].disk.hash
self._caches = {}
self._deques = {}
self._indexes = {}
def __init__(self, cache=None, expire=None):
if isinstance(cache, Cache):
pass
elif cache is None:
cache = Cache(tempfile.mkdtemp())
else:
cache = Cache(cache)
self._cache = cache
self._expire = expire
def __init__(self, cache=None, expire=None):
if isinstance(cache, Cache):
pass
elif cache is None:
cache = Cache(tempfile.mkdtemp())
else:
cache = Cache(cache)
self._cache = cache
self._expire = expire
"Ordered cache to save keys in order."
import sqlite3
from .core import Cache, Disk, ENOVAL
class OrderedCache(Cache):
"Cache that saves keys in order."
def __init__(self, directory, timeout=60, disk=Disk(),
**settings):
"""Initialize OrderedCache instance.
:param str directory: cache directory
:param float timeout: SQLite connection timeout
:param disk: `Disk`: instance for serialization
:param settings: any of `DEFAULT_SETTINGS`
"""
super(self.__class__, self).__init__(directory, timeout, disk)
def first(self, default=ENOVAL):
"""Retrieve the first key in the cache based on age