]>
Commit | Line | Data |
---|---|---|
11fdf7f2 TL |
1 | """ |
2 | This is a minimal implementation of TTL-ed lru_cache function. | |
3 | ||
4 | Based on Python 3 functools and backports.functools_lru_cache. | |
5 | """ | |
11fdf7f2 | 6 | |
11fdf7f2 | 7 | from collections import OrderedDict |
f67539c2 | 8 | from functools import wraps |
11fdf7f2 TL |
9 | from threading import RLock |
10 | from time import time | |
11 | ||
9f95a23c TL |
12 | try: |
13 | from typing import Tuple | |
14 | except ImportError: | |
15 | pass # For typing only | |
16 | ||
11fdf7f2 TL |
17 | |
18 | def ttl_cache(ttl, maxsize=128, typed=False): | |
19 | if typed is not False: | |
20 | raise NotImplementedError("typed caching not supported") | |
21 | ||
22 | def decorating_function(function): | |
9f95a23c | 23 | cache = OrderedDict() # type: OrderedDict[object, Tuple[bool, float]] |
11fdf7f2 TL |
24 | stats = [0, 0, 0] |
25 | rlock = RLock() | |
9f95a23c TL |
26 | setattr(function, 'cache_info', lambda: |
27 | "hits={}, misses={}, expired={}, maxsize={}, currsize={}".format( | |
28 | stats[0], stats[1], stats[2], maxsize, len(cache))) | |
11fdf7f2 TL |
29 | |
30 | @wraps(function) | |
31 | def wrapper(*args, **kwargs): | |
32 | key = args + tuple(kwargs.items()) | |
33 | with rlock: | |
34 | refresh = True | |
35 | if key in cache: | |
36 | (ret, ts) = cache[key] | |
37 | del cache[key] | |
38 | if time() - ts < ttl: | |
39 | refresh = False | |
40 | stats[0] += 1 | |
41 | else: | |
42 | stats[2] += 1 | |
43 | ||
44 | if refresh: | |
45 | ret = function(*args, **kwargs) | |
46 | ts = time() | |
47 | if len(cache) == maxsize: | |
48 | cache.popitem(last=False) | |
49 | stats[1] += 1 | |
50 | ||
51 | cache[key] = (ret, ts) | |
52 | ||
53 | return ret | |
54 | ||
55 | return wrapper | |
56 | return decorating_function |