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
|
"""
This is a minimal implementation of TTL-ed lru_cache function.
Based on Python 3 functools and backports.functools_lru_cache.
"""
from __future__ import absolute_import
from collections import OrderedDict
from functools import wraps
from threading import RLock
from time import time
try:
from typing import Tuple
except ImportError:
pass # For typing only
def ttl_cache(ttl, maxsize=128, typed=False):
if typed is not False:
raise NotImplementedError("typed caching not supported")
def decorating_function(function):
cache = OrderedDict() # type: OrderedDict[object, Tuple[bool, float]]
stats = [0, 0, 0]
rlock = RLock()
setattr(function, 'cache_info', lambda:
"hits={}, misses={}, expired={}, maxsize={}, currsize={}".format(
stats[0], stats[1], stats[2], maxsize, len(cache)))
@wraps(function)
def wrapper(*args, **kwargs):
key = args + tuple(kwargs.items())
with rlock:
refresh = True
if key in cache:
(ret, ts) = cache[key]
del cache[key]
if time() - ts < ttl:
refresh = False
stats[0] += 1
else:
stats[2] += 1
if refresh:
ret = function(*args, **kwargs)
ts = time()
if len(cache) == maxsize:
cache.popitem(last=False)
stats[1] += 1
cache[key] = (ret, ts)
return ret
return wrapper
return decorating_function
|
