expiring-dict.py

K = typing.TypeVar('K')
V = typing.TypeVar('V')

class ExpiringCache(typing.Generic[K, V]):
    def __init__(self, cache_name: str, max_size: int, age_params: dict):
        self.entries: OrderedDict[typing.Any, typing.Any] = OrderedDict()
        self.cache_name = cache_name
        self.age_params = age_params
        self.max_size = max_size

        self.metric_cache_write = Counter(
            'expiring_dict_cache_write',
            'Total number of inserts.',
            [ 'cache_name' ]
        )
        self.metric_cache_get = Counter(
            'expiring_dict_cache_get',
            'Count of cache hits and misses.',
            [ 'cache_name', 'result' ]
        )
        self.metric_cache_expirations = Counter(
            'expiring_dict_cache_expirations',
            'Count of cache expirations.',
            [ 'cache_name' ]
        )
        self.metric_cache_overflow = Counter(
            'expiring_dict_cache_overflow',
            'Total count of the number of cache evictions based on max size',
            [ 'cache_name', 'size' ]
        )


    def get(self, key: K) -> typing.Optional[V]:
        record = self.entries.get(key)

        if record is None:
            self.metric_cache_get.labels(self.cache_name, 'miss').inc()
            return None

        if record['validity'] < pendulum.now():
            self.metric_cache_get.labels(self.cache_name, 'miss').inc()
            self.metric_cache_expirations.labels(self.cache_name).inc()
            # Delete the expired entry
            del self.entries[key]
            return None

        self.metric_cache_get.labels(self.cache_name, 'hit').inc()
        return record['value']

    def insert(self, key: K, value: V):
        if len(self.entries) >= self.max_size:
            self.metric_cache_overflow.labels(self.cache_name, self.max_size).inc()
            # Delete the most unused item
            self.entries.popitem(last=True)

        self.entries[key] = {
            'validity': pendulum.now().add(**self.age_params),
            'value': value
        }
        self.entries.move_to_end(key, last=False)
        self.metric_cache_write.labels(self.cache_name).inc()