/usr/lib/python3/dist-packages/zict/lru.py is in python3-zict 0.1.3-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | from __future__ import absolute_import, division, print_function
from heapdict import heapdict
from .common import ZictBase, close
def do_nothing(k, v):
pass
class LRU(ZictBase):
""" Evict Least Recently Used Elements
Parameters
----------
n: int
Number of elements to keep, or total weight if weight= is used
d: MutableMapping
Dictionary in which to hold elements
on_evict: list of callables
Function:: k, v -> action to call on key value pairs prior to eviction
weight: callable
Function:: k, v -> number to determine the size of keeping the item in
the mapping. Defaults to ``(k, v) -> 1``
Examples
--------
>>> lru = LRU(2, dict(), on_evict=lambda k, v: print("Lost", k, v))
>>> lru['x'] = 1
>>> lru['y'] = 2
>>> lru['z'] = 3
Lost x 1
"""
def __init__(self, n, d, on_evict=None, weight=lambda k, v: 1):
self.d = d
self.n = n
self.heap = heapdict()
self.i = 0
if callable(on_evict):
on_evict = [on_evict]
self.on_evict = on_evict or []
self.weight = weight
self.total_weight = 0
self.weights = dict()
def __getitem__(self, key):
result = self.d[key]
self.i += 1
self.heap[key] = self.i
return result
def __setitem__(self, key, value):
if key in self.d:
del self[key]
weight = self.weight(key, value)
if weight <= self.n:
self.d[key] = value
self.i += 1
self.heap[key] = self.i
self.weights[key] = weight
self.total_weight += weight
else:
for cb in self.on_evict:
cb(key, value)
while self.total_weight > self.n:
self.evict()
def evict(self):
""" Evict least recently used key
This is typically called from internal use, but can be externally
triggered as well.
Returns
-------
k: key
v: value
w: weight
"""
k, priority = self.heap.popitem()
weight = self.weights.pop(k)
self.total_weight -= weight
v = self.d.pop(k)
for cb in self.on_evict:
cb(k, v)
return k, v, weight
def __delitem__(self, key):
del self.d[key]
del self.heap[key]
self.total_weight -= self.weights.pop(key)
def keys(self):
return self.d.keys()
def values(self):
return self.d.values()
def items(self):
return self.d.items()
def __len__(self):
return len(self.d)
def __iter__(self):
return iter(self.d)
def __contains__(self, key):
return key in self.d
def __str__(self):
sub = str(self.d) if not isinstance(self.d, dict) else 'dict'
return '<LRU: %s/%s on %s>' % (self.total_weight, self.n, sub)
__repr__ = __str__
def flush(self):
self.d.flush()
def close(self):
close(self.d)
|