import operator as op from collections import OrderedDict from collections.abc import (
ItemsView,
KeysView,
MutableMapping,
Sequence,
ValuesView,
) from contextlib import contextmanager from shutil import rmtree
from .core import ENOVAL, Cache
def _make_compare(seq_op, doc): """Make compare method with Sequence semantics."""
def compare(self, that): """Compare method for deque and sequence.""" ifnot isinstance(that, Sequence): return NotImplemented
len_self = len(self)
len_that = len(that)
if len_self != len_that: if seq_op is op.eq: returnFalse if seq_op is op.ne: returnTrue
for alpha, beta in zip(self, that): if alpha != beta: return seq_op(alpha, beta)
return seq_op(len_self, len_that)
compare.__name__ = '__{0}__'.format(seq_op.__name__)
doc_str = 'Return True if and only if deque is {0} `that`.'
compare.__doc__ = doc_str.format(doc)
return compare
class Deque(Sequence): """Persistent sequence with double-ended queue semantics.
Double-ended queue is an ordered collection with optimized access at its
endpoints.
Items are serialized to disk. Deque may be initialized from directory path
where items are stored.
if index >= 0: if index >= len_self: raise IndexError('deque index out of range')
for key in self._cache.iterkeys(): if index == 0: try: return func(key) except KeyError: continue
index -= 1 else: if index < -len_self: raise IndexError('deque index out of range')
index += 1
for key in self._cache.iterkeys(reverse=True): if index == 0: try: return func(key) except KeyError: continue
index += 1
def copy(self): """Copy deque with same directory and max length."""
TypeSelf = type(self) return TypeSelf(directory=self.directory, maxlen=self.maxlen)
def count(self, value): """Return number of occurrences of `value` in deque.
>>> deque = Deque()
>>> deque += [num for num in range(1, 5) for _ in range(num)]
>>> deque.count(0)
0
>>> deque.count(1)
1
>>> deque.count(4)
4
:param value: value to count in deque
:return: count of items equal to value in deque
""" return sum(1 for item in self if value == item)
def extend(self, iterable): """Extend back side of deque with values from `iterable`.
:param iterable: iterable of values
""" for value in iterable:
self._append(value)
_extend = extend
def extendleft(self, iterable): """Extend front side of deque with value from `iterable`.
:return: value at back of deque
:raises IndexError: if deque is empty
"""
default = None, ENOVAL
_, value = self._cache.peek(default=default, side='back', retry=True) if value is ENOVAL: raise IndexError('peek from an empty deque') return value
def peekleft(self): """Peek at value at front of deque.
:return: value at front of deque
:raises IndexError: if deque is empty
"""
default = None, ENOVAL
_, value = self._cache.peek(default=default, side='front', retry=True) if value is ENOVAL: raise IndexError('peek from an empty deque') return value
def pop(self): """Remove and return value at back of deque.
If deque is empty then raise IndexError.
>>> deque = Deque()
>>> deque += 'ab'
>>> deque.pop() 'b'
>>> deque.pop() 'a'
>>> deque.pop()
Traceback (most recent call last):
...
IndexError: pop from an empty deque
:return: value at back of deque
:raises IndexError: if deque is empty
"""
default = None, ENOVAL
_, value = self._cache.pull(default=default, side='back', retry=True) if value is ENOVAL: raise IndexError('pop from an empty deque') return value
_pop = pop
def popleft(self): """Remove and return value at front of deque.
>>> deque = Deque()
>>> deque += 'ab'
>>> deque.popleft() 'a'
>>> deque.popleft() 'b'
>>> deque.popleft()
Traceback (most recent call last):
...
IndexError: pop from an empty deque
:return: value at front of deque
:raises IndexError: if deque is empty
"""
default = None, ENOVAL
_, value = self._cache.pull(default=default, retry=True) if value is ENOVAL: raise IndexError('pop from an empty deque') return value
_popleft = popleft
def remove(self, value): """Remove first occurrence of `value` in deque.
:param value: value to remove
:raises ValueError: if value notin deque
"""
_cache = self._cache
for key in _cache.iterkeys(): try:
item = _cache[key] except KeyError: continue else: if value == item: try: del _cache[key] except KeyError: continue return
raise ValueError('deque.remove(value): value not in deque')
""" # pylint: disable=protected-access # GrantJ 2019-03-22 Consider using an algorithm that swaps the values # at two keys. Like self._cache.swap(key1, key2, retry=True) The swap # method would exchange the values at two given keys. Then, using a # forward iterator and a reverse iterator, the reverse method could # avoid making copies of the values.
temp = Deque(iterable=reversed(self))
self._clear()
self._extend(temp)
directory = temp.directory
temp._cache.close() del temp
rmtree(directory)
def rotate(self, steps=1): """Rotate deque right by `steps`.
for _ in range(steps): try:
value = self._pop() except IndexError: return else:
self._appendleft(value) else:
steps *= -1
steps %= len_self
for _ in range(steps): try:
value = self._popleft() except IndexError: return else:
self._append(value)
__hash__ = None# type: ignore
@contextmanager def transact(self): """Context manager to perform a transaction by locking the deque.
While the deque is locked, no other write operation is permitted.
Transactions should therefore be as short as possible. Read and write
operations performed in a transaction are atomic. Read operations may
occur concurrent to a transaction.
Transactions may be nested and may not be shared between threads.
def peekitem(self, last=True): """Peek at key and value item pair in index based on iteration order.
>>> index = Index()
>>> for num, letter in enumerate('xyz'):
... index[letter] = num
>>> index.peekitem()
('z', 2)
>>> index.peekitem(last=False)
('x', 0)
:param bool last: last item in iteration order (default True)
:return: key and value item pair
:raises KeyError: if cache is empty
""" return self._cache.peekitem(last, retry=True)
def pop(self, key, default=ENOVAL): """Remove corresponding item for `key` from index and return value.
If `key` is missing then return `default`. If `default` is `ENOVAL`
then raise KeyError.
:param key: key for item
:param default: return value if key is missing (default ENOVAL)
:return: value for item if key is found else default
:raises KeyError: if key isnot found and default is ENOVAL
"""
_cache = self._cache
value = _cache.pop(key, default=default, retry=True) if value is ENOVAL: raise KeyError(key) return value
def popitem(self, last=True): """Remove and return item pair.
Item pairs are returned in last-in-first-out (LIFO) order if last is Trueelse first-in-first-out (FIFO) order. LIFO order imitates a stack and FIFO order imitates a queue.
:param value: value for item
:param str prefix: key prefix (default None, key is integer)
:param str side: either 'back'or'front' (default 'back')
:return: key for item in cache
def pull(self, prefix=None, default=(None, None), side='front'): """Pull key and value item pair from `side` of queue in index.
When prefix isNone, integer keys are used. Otherwise, string keys are
used in the format "prefix-integer". Integer starts at 500 trillion.
If queue is empty, return default.
Defaults to pulling key and value item pairs from front of queue. Set
side to 'back' to pull from back of queue. Side must be one of 'front' or'back'.
See also `Index.push`.
>>> index = Index()
>>> for letter in'abc':
... print(index.push(letter))
500000000000000
500000000000001
500000000000002
>>> key, value = index.pull()
>>> print(key)
500000000000000
>>> value 'a'
>>> _, value = index.pull(side='back')
>>> value 'c'
>>> index.pull(prefix='fruit')
(None, None)
:param str prefix: key prefix (default None, key is integer)
:param default: value to returnif key is missing
(default (None, None))
:param str side: either 'front'or'back' (default 'front')
:return: key and value item pair or default if queue is empty
def __eq__(self, other): """index.__eq__(other) <==> index == other
Compare equality for index and `other`.
Comparison to another index or ordered dictionary is
order-sensitive. Comparison to all other mappings is order-insensitive.
>>> index = Index()
>>> pairs = [('a', 1), ('b', 2), ('c', 3)]
>>> index.update(pairs)
>>> from collections import OrderedDict
>>> od = OrderedDict(pairs)
>>> index == od True
>>> index == {'c': 3, 'b': 2, 'a': 1} True
:param other: other mapping in equality comparison
:return: Trueif index equals other
""" if len(self) != len(other): returnFalse
if isinstance(other, (Index, OrderedDict)):
alpha = ((key, self[key]) for key in self)
beta = ((key, other[key]) for key in other)
pairs = zip(alpha, beta) returnnot any(a != x or b != y for (a, b), (x, y) in pairs) else: return all(self[key] == other.get(key, ENOVAL) for key in self)
def __ne__(self, other): """index.__ne__(other) <==> index != other
Compare inequality for index and `other`.
Comparison to another index or ordered dictionary is
order-sensitive. Comparison to all other mappings is order-insensitive.
>>> index = Index()
>>> index.update([('a', 1), ('b', 2), ('c', 3)])
>>> from collections import OrderedDict
>>> od = OrderedDict([('c', 3), ('b', 2), ('a', 1)])
>>> index != od True
>>> index != {'a': 1, 'b': 2} True
:param other: other mapping in inequality comparison
:return: Trueif index does not equal other
Decorator to wrap callable with memoizing function using cache.
Repeated calls with the same arguments will lookup result in cache and
avoid function evaluation.
If name is set to None (default), the callable name will be determined
automatically.
If typed is set to True, function arguments of different types will be
cached separately. For example, f(3) and f(3.0) will be treated as
distinct calls with distinct results.
The original underlying function is accessible through the __wrapped__
attribute. This is useful for introspection, for bypassing the cache, orfor rewrapping the function with a different cache.
>>> from diskcache import Index
>>> mapping = Index()
>>> @mapping.memoize()
... def fibonacci(number):
... if number == 0:
... return 0
... elif number == 1:
... return 1
... else:
... return fibonacci(number - 1) + fibonacci(number - 2)
>>> print(fibonacci(100))
354224848179261915075
An additional `__cache_key__` attribute can be used to generate the
cache key used for the given arguments.
Remember to call memoize when decorating a callable. If you forget,
then a TypeError will occur. Note the lack of parenthenses after
memoize below:
>>> @mapping.memoize
... def test():
... pass
Traceback (most recent call last):
...
TypeError: name cannot be callable
:param str name: name given for callable (default None, automatic)
:param bool typed: cache different types separately (default False)
:param set ignore: positional or keyword args to ignore (default ())
:return: callable decorator
@contextmanager def transact(self): """Context manager to perform a transaction by locking the index.
While the index is locked, no other write operation is permitted.
Transactions should therefore be as short as possible. Read and write
operations performed in a transaction are atomic. Read operations may
occur concurrent to a transaction.
Transactions may be nested and may not be shared between threads.
>>> from diskcache import Index
>>> mapping = Index()
>>> with mapping.transact(): # Atomically increment two keys.
... mapping['total'] = mapping.get('total', 0) + 123.4
... mapping['count'] = mapping.get('count', 0) + 1
>>> with mapping.transact(): # Atomically calculate average.
... average = mapping['total'] / mapping['count']
>>> average
123.4
:return: context manager for use in `with` statement
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.