Source code for sunpy.database.caching

# Author: Simon Liedtke <liedtke.simon@googlemail.com>
#
# This module was developed with funding provided by
# the Google Summer of Code (2013).

from abc import ABCMeta, abstractmethod, abstractproperty
from collections import Counter, OrderedDict
from collections.abc import MutableMapping

__all__ = ['BaseCache', 'LRUCache', 'LFUCache']


[docs]class BaseCache(metaclass=ABCMeta): """ BaseCache is a class that saves and operates on an OrderedDict. It has a certain capacity, stored in the attribute ``maxsize``. Whether this capacity is reached, can be checked by using the boolean property `is_full`. To implement a custom cache, inherit from this class and override the methods ``__getitem__`` and ``__setitem__``. Call the method `sunpy.database.caching.BaseCache.callback` as soon as an item from the cache is removed. """ def __init__(self, maxsize=float('inf')): self.maxsize = maxsize self._dict = OrderedDict()
[docs] def get(self, key, default=None): """Return the corresponding value to ``key`` if ``key`` is in the cache, ``default`` otherwise. This method has no side-effects, multiple calls with the same cache and the same passed key must always return the same value. """ try: return self._dict[key] except KeyError: return default
@abstractmethod def __getitem__(self, key): """abstract method: this method must be overwritten by inheriting subclasses. It defines what happens if an item from the cache is attempted to be accessed. """ return @abstractmethod def __setitem__(self, key, value): """abstract method: this method must be overwritten by inheriting subclasses. It defines what happens if a new value should be assigned to the given key. If the given key does already exist in the cache or not must be checked by the person who implements this method. """ @abstractproperty def to_be_removed(self): """The item that will be removed on the next :meth:`sunpy.database.caching.BaseCache.remove` call. """
[docs] @abstractmethod def remove(self): """Call this method to manually remove one item from the cache. Which item is removed, depends on the implementation of the cache. After the item has been removed, the callback method is called. """
[docs] def callback(self, key, value): """This method should be called (by convention) if an item is removed from the cache because it is full. The passed key and value are the ones that are removed. By default this method does nothing, but it can be customized in a custom cache that inherits from this base class. """
@property def is_full(self): """True if the number of items in the cache equals ``self.maxsize``, False otherwise. """ return len(self._dict) == self.maxsize def __delitem__(self, key): self._dict.__delitem__(key) def __contains__(self, key): return key in self._dict.keys() def __len__(self): return len(self._dict) def __iter__(self): yield from self._dict.__iter__() def __reversed__(self): yield from self._dict.__reversed__()
[docs] def clear(self): return self._dict.clear()
[docs] def keys(self): return list(self._dict.keys())
[docs] def values(self): return list(self._dict.values())
[docs] def items(self): return list(self._dict.items())
[docs] def iterkeys(self): return iter(self._dict.keys())
[docs] def itervalues(self): yield from self._dict.values()
[docs] def iteritems(self): yield from self._dict.items()
[docs] def update(self, *args, **kwds): self._dict.update(*args, **kwds)
[docs] def pop(self, key, default=MutableMapping._MutableMapping__marker): return self._dict.pop(key, default)
[docs] def setdefault(self, key, default=None): return self._dict.setdefault(key, default)
[docs] def popitem(self, last=True): return self._dict.popitem(last)
def __reduce__(self): return self._dict.__reduce__()
[docs] def copy(self): return self._dict.copy()
def __eq__(self, other): return self._dict.__eq__(other) def __ne__(self, other): return self._dict.__ne__(other)
[docs] def viewkeys(self): return self._dict.keys()
[docs] def viewvalues(self): return self._dict.values()
[docs] def viewitems(self): return self._dict.items()
[docs] @classmethod def fromkeys(cls, iterable, value=None): return OrderedDict.fromkeys(iterable, value)
def __repr__(self): return '{}({!r})'.format(self.__class__.__name__, dict(self._dict))
[docs]class LRUCache(BaseCache): """ LRUCache """ @property def to_be_removed(self): """Return the least recently used key and its corresponding value as a tuple. """ return next(self.iteritems())
[docs] def remove(self): """Remove the least recently used item.""" self.callback(*self.popitem(last=False))
def __getitem__(self, key): """Returns the value which is associated to the given key and put it with its associated value to the end of this cache. Raises ------ KeyError If the key cannot be found in the cache. """ if key in self: value = self._dict.__getitem__(key) del self[key] self._dict.__setitem__(key, value) return value raise KeyError def __setitem__(self, key, value): """If the key does already exist in the cache, move it to the end of this cache. Otherwise, set a new value and put it to the end of this cache. If the cache is full, remove the least recently used item before inserting the new key-value pair. """ if key in self: del self[key] if self.is_full: self.remove() self._dict.__setitem__(key, value)
[docs]class LFUCache(BaseCache): """ LFUCache """ def __init__(self, maxsize=float('inf')): self.usage_counter = Counter() BaseCache.__init__(self, maxsize) @property def to_be_removed(self): """Returns the key with the lowest times of access and its corresponding value as a tuple. """ min_ = float('inf') lfu_key = None for k, v in self.usage_counter.items(): if v < min_: min_ = v lfu_key = k return lfu_key, self.get(lfu_key)
[docs] def remove(self): """Remove the least frequently used item.""" lfu_key, val = self.to_be_removed del self[lfu_key] del self.usage_counter[lfu_key] self.callback(lfu_key, val)
def __getitem__(self, key): """Returns the value which is associated to the given key and increments the frequency counter of this key. Raises ------ KeyError If the key cannot be found in the cache. """ value = self._dict.__getitem__(key) self.usage_counter[key] += 1 return value def __setitem__(self, key, value): """Increment the frequency counter of the given key if it is already present in the cache, otherwise set it to 1. If the cache is full, remove the least frequently used item before inserting the new key-value pair. """ self.usage_counter[key] += 1 if self.is_full: self.remove() self._dict.__setitem__(key, value)