import collections import functools from itertools import ifilterfalse
def lru_cache(maxsize=100, on_purge=None): """Least-recently-used cache decorator. Arguments to the cached function must be hashable. Clear the cache with f.clear(). """ maxqueue = maxsize * 10
@functools.wraps(user_function) def wrapper(*args, **kwargs): # cache key records both positional and keyword args key = args if kwargs: key += (kwd_mark,) + tuple(sorted(kwargs.items()))
# record recent use of this key queue_append(key) refcount[key] += 1
# get cache entry or compute if not found try: result = cache[key] except KeyError: result = user_function(*args, **kwargs) cache[key] = result
# purge least recently used cache entry if len(cache) > maxsize: key = queue_popleft() refcount[key] -= 1 while refcount[key]: key = queue_popleft() refcount[key] -= 1 if on_purge: on_purge(cache[key]) del cache[key], refcount[key]
# periodically compact the queue by eliminating duplicate keys # while preserving order of most recent access if len(queue) > maxqueue: refcount.clear() queue_appendleft(sentinel) for key in ifilterfalse(refcount.__contains__, iter(queue_pop, sentinel)): queue_appendleft(key) refcount[key] = 1 return result
def clear(): if on_purge: for value in cache.itervalues(): on_purge(value) cache.clear() queue.clear() refcount.clear()