Source code for aioxmpp.cache
########################################################################
# File name: cache.py
# This file is part of: aioxmpp
#
# LICENSE
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this program. If not, see
# <http://www.gnu.org/licenses/>.
#
########################################################################
"""
:mod:`~aioxmpp.cache` --- Utilities for implementing caches
###########################################################
.. versionadded:: 0.9
This module was added in version 0.9.
.. autoclass:: LRUDict
"""
import collections.abc
class Node:
__slots__ = ("prev", "next_", "key", "value")
def _init_linked_list():
root = Node()
root.prev = root
root.next_ = root
root.key = None
root.value = None
return root
def _remove_node(node):
node.next_.prev = node.prev
node.prev.next_ = node.next_
return node
def _insert_node(before, new_node):
new_node.next_ = before.next_
new_node.next_.prev = new_node
new_node.prev = before
before.next_ = new_node
def _length(node):
# this is used only for testing
cur = node.next_
i = 0
while cur is not node:
i += 1
cur = cur.next_
return i
def _has_consistent_links(node, node_dict=None):
# this is used only for testing
cur = node.next_
if cur.prev is not node:
return False
while cur is not node:
if node_dict is not None and node_dict[cur.key] is not cur:
return False
if cur is not cur.next_.prev:
return False
cur = cur.next_
return True
[docs]class LRUDict(collections.abc.MutableMapping):
"""
Size-restricted dictionary with Least Recently Used expiry policy.
.. versionadded:: 0.9
The :class:`LRUDict` supports normal dictionary-style access and implements
:class:`collections.abc.MutableMapping`.
When the :attr:`maxsize` is exceeded, as many entries as needed to get
below the :attr:`maxsize` are removed from the dict. Least recently used
entries are purged first. Setting an entry does *not* count as use!
.. autoattribute:: maxsize
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.__links = {}
self.__root = _init_linked_list()
self.__maxsize = 1
def _test_consistency(self):
"""
This method is only used for testing to assert that the operations
leave the LRUDict in a valid state.
"""
return (_length(self.__root) == len(self.__links) and
_has_consistent_links(self.__root, self.__links))
def _purge(self):
if self.__maxsize is None:
return
while len(self.__links) > self.__maxsize:
link = _remove_node(self.__root.prev)
del self.__links[link.key]
@property
def maxsize(self):
"""
Maximum size of the cache. Changing this property purges overhanging
entries immediately.
If set to :data:`None`, no limit on the number of entries is imposed.
Do **not** use a limit of :data:`None` for data where the `key` is
under control of a remote entity.
Use cases for :data:`None` are those where you only need the explicit
expiry feature, but not the LRU feature.
"""
return self.__maxsize
@maxsize.setter
def maxsize(self, value):
if value is not None and value <= 0:
raise ValueError("maxsize must be positive integer or None")
self.__maxsize = value
self._purge()
def __len__(self):
return len(self.__links)
def __iter__(self):
return iter(self.__links)
def __setitem__(self, key, value):
try:
self.__links[key].value = value
except KeyError:
link = Node()
link.key = key
link.value = value
self.__links[key] = link
_insert_node(self.__root, link)
self._purge()
def __getitem__(self, key):
link = self.__links[key]
_remove_node(link)
_insert_node(self.__root, link)
return link.value
def __delitem__(self, key):
link = self.__links.pop(key)
_remove_node(link)
def clear(self):
self.__links.clear()
self.__root = _init_linked_list()