89 lines
1.7 KiB
Python
89 lines
1.7 KiB
Python
'''Simple caches that uses ordered dicts'''
|
|
|
|
import re
|
|
|
|
from datetime import datetime
|
|
from collections import OrderedDict
|
|
|
|
|
|
def parse_ttl(ttl):
|
|
m = re.match(r'^(\d+)([smhdw]?)$', ttl)
|
|
|
|
if not m:
|
|
raise ValueError(f'Invalid TTL length: {ttl}')
|
|
|
|
amount = m.group(1)
|
|
unit = m.group(2)
|
|
|
|
if not unit:
|
|
raise ValueError('Missing numerical length in TTL')
|
|
|
|
units = {
|
|
's': 1,
|
|
'm': 60,
|
|
'h': 60 * 60,
|
|
'd': 24 * 60 * 60,
|
|
'w': 7 * 24 * 60 * 60
|
|
}
|
|
|
|
multiplier = units.get(unit)
|
|
|
|
if not multiplier:
|
|
raise ValueError(f'Invalid time unit: {unit}')
|
|
|
|
return multiplier * int(amount)
|
|
|
|
|
|
class TTLCache(OrderedDict):
|
|
def __init__(self, maxsize=1024, ttl='1h'):
|
|
self.ttl = parse_ttl(ttl)
|
|
self.maxsize = maxsize
|
|
|
|
def remove(self, key):
|
|
if self.get(key):
|
|
del self[key]
|
|
|
|
def store(self, key, value):
|
|
timestamp = int(datetime.timestamp(datetime.now()))
|
|
item = self.get(key)
|
|
|
|
while len(self) >= self.maxsize and self.maxsize != 0:
|
|
self.popitem(last=False)
|
|
|
|
self[key] = {'data': value, 'timestamp': timestamp + self.ttl}
|
|
self.move_to_end(key)
|
|
|
|
def fetch(self, key):
|
|
item = self.get(key)
|
|
timestamp = int(datetime.timestamp(datetime.now()))
|
|
|
|
if not item:
|
|
return
|
|
|
|
if timestamp >= self[key]['timestamp']:
|
|
del self[key]
|
|
return
|
|
|
|
self[key]['timestamp'] = timestamp + self.ttl
|
|
self.move_to_end(key)
|
|
return self[key]['data']
|
|
|
|
|
|
class LRUCache(OrderedDict):
|
|
def __init__(self, maxsize=1024):
|
|
self.maxsize = maxsize
|
|
|
|
def remove(self, key):
|
|
if key in self:
|
|
del self[key]
|
|
|
|
def store(self, key, value):
|
|
while len(self) >= self.maxsize and self.maxsize != 0:
|
|
self.popitem(last=False)
|
|
|
|
self[key] = value
|
|
self.move_to_end(key)
|
|
|
|
def fetch(self, key):
|
|
return self.get(key)
|