Source code for

import time
from itertools import takewhile
import operator
from collections import OrderedDict
from abc import abstractmethod, ABC

[docs]class IStorage(ABC): """ Local storage for this node. IStorage implementations of get must return the same type as put in by set """
[docs] @abstractmethod def __setitem__(self, key, value): """ Set a key to the given value. """
[docs] @abstractmethod def __getitem__(self, key): """ Get the given key. If item doesn't exist, raises C{KeyError} """
[docs] @abstractmethod def get(self, key, default=None): """ Get given key. If not found, return default. """
[docs] @abstractmethod def iter_older_than(self, seconds_old): """ Return the an iterator over (key, value) tuples for items older than the given secondsOld. """
@abstractmethod def __iter__(self): """ Get the iterator for this storage, should yield tuple of (key, value) """
[docs]class ForgetfulStorage(IStorage): def __init__(self, ttl=604800): """ By default, max age is a week. """ = OrderedDict() self.ttl = ttl
[docs] def __setitem__(self, key, value): if key in del[key][key] = (time.monotonic(), value) self.cull()
[docs] def cull(self): for _, _ in self.iter_older_than(self.ttl):
[docs] def get(self, key, default=None): self.cull() if key in return self[key] return default
[docs] def __getitem__(self, key): self.cull() return[key][1]
def __repr__(self): self.cull() return repr(
[docs] def iter_older_than(self, seconds_old): min_birthday = time.monotonic() - seconds_old zipped = self._triple_iter() matches = takewhile(lambda r: min_birthday >= r[1], zipped) return list(map(operator.itemgetter(0, 2), matches))
def _triple_iter(self): ikeys = ibirthday = map(operator.itemgetter(0), ivalues = map(operator.itemgetter(1), return zip(ikeys, ibirthday, ivalues) def __iter__(self): self.cull() ikeys = ivalues = map(operator.itemgetter(1), return zip(ikeys, ivalues)