timedcache.py

import collections
import dataclasses
import functools
import time
import typing as T

NumberType = T.Union[int, float]
X = T.TypeVar("X")


@dataclasses.dataclass
class _TimedCacheEntry(T.Generic[X]):
    value: X
    expires_at: float


def timed_cache(seconds: NumberType = 60) -> T.Callable[[T.Callable[..., X]], T.Callable[..., X]]:
    def _timed_cache(function: T.Callable[..., X]) -> T.Callable[..., X]:
        cache: collections.OrderedDict[T.Hashable, _TimedCacheEntry[X]] = collections.OrderedDict()

        @functools.wraps(function)
        def wrapped(*args: T.Hashable, **kwargs: T.Hashable) -> X:
            now = time.monotonic()
            for key, entry in tuple(cache.items()):
                if entry.expires_at >= now:
                    break

                del cache[key]

            key = args, tuple(kwargs.items())
            if key in cache:
                return cache[key].value

            result = function(*args, **kwargs)
            cache[key] = _TimedCacheEntry(result, now + seconds)
            return result

        return wrapped

    return _timed_cache