Welcome, guest | Sign In | My Account | Store | Cart

For most cases where you want to store a limited amount of data, functools.partial is sufficient. However, if you want or need more control over your data, especially specifying an expiration date for your entries, this can come in handy.

Read the docstrings for implementation details.

Python, 104 lines
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
'''A simple implementation of a LRU dict that supports discarding by maximum
capacity and by maximum time not being used.'''

from collections import OrderedDict
import time

class LRUDict(OrderedDict):
    '''An dict that can discard least-recently-used items, either by maximum capacity
    or by time to live.
    An item's ttl is refreshed (aka the item is considered "used") by direct access
    via [] or get() only, not via iterating over the whole collection with items()
    for example.
    Expired entries only get purged after insertions or changes. Either call purge()
    manually or check an item's ttl with ttl() if that's unacceptable.
    '''
    def __init__(self, *args, maxduration=None, maxsize=128, **kwargs):
        '''Same arguments as OrderedDict with these 2 additions:
        maxduration: number of seconds entries are kept. 0 or None means no timelimit.
        maxsize: maximum number of entries being kept.'''
        super().__init__(*args, **kwargs)
        self.maxduration = maxduration
        self.maxsize = maxsize
        self.purge()

    def purge(self):
        """Removes expired or overflowing entries."""
        if self.maxsize:
            # pop until maximum capacity is reached
            overflowing = max(0, len(self) - self.maxsize)
            for _ in range(overflowing):
                self.popitem(last=False)
        if self.maxduration:
            # expiration limit
            limit = time.time() - self.maxduration
            # as long as there are still items in the dictionary
            while self:
                # look at the oldest (front)
                _, lru = next(iter(super().values()))
                # if it is within the timelimit, we're fine
                if lru > limit:
                    break
                # otherwise continue to pop the front
                self.popitem(last=False)

    def __getitem__(self, key):
        # retrieve item
        value = super().__getitem__(key)[0]
        # update lru time
        super().__setitem__(key, (value, time.time()))
        self.move_to_end(key)
        return value

    def get(self, key, default=None):
        try:
            return self[key]
        except KeyError:
            return default

    def ttl(self, key):
        '''Returns the number of seconds this item will live.
        The item might still be deleted if maxsize is reached.
        The time to live can be negative, as for expired items
        that have not been purged yet.'''
        if self.maxduration:
            lru = super().__getitem__(key)[1]
            return self.maxduration - (time.time() - lru)

    def __setitem__(self, key, value):
        super().__setitem__(key, (value, time.time()))
        self.purge()
        
    def items(self):
        # remove ttl from values
        return ((k, v) for k, (v, _) in super().items())
    
    def values(self):
        # remove ttl from values
        return (v for v, _ in super().values())


def main():
    dct = LRUDict(maxduration=2)
    print(dct)  # empty
    dct["a"] = 5
    time.sleep(1)
    print(dct)  # a
    dct["b"] = 10
    time.sleep(1.5)
    print(dct)  # a, b
    dct["c"] = 20
    print(dct)  # b, c
    print(dct.get("a"))
    print(dct["b"])
    print(dct["c"])
    time.sleep(1)
    dct.purge()
    print(dct)  # c
    for k, v in dct.items():
        print("k:%s, v:%s" % (k, v))
    for v in dct.values():
        print("v:%s" % (v, ))

if __name__ == "__main__":
    main()

When the functionality of a fixed ttl is not needed, in most cases one could use a soluton like this:

class SomeRepository:
    def __init__(self, hashable_data):
        self.hashable_data = hashable_data
    @functools.lru_cache()
    def __getitem__(self, key):
        return Widget(key, self.hashable_data)