Skip to content

Instantly share code, notes, and snippets.

@cryzed

cryzed/timedcache.py

Last active May 23, 2021 21:47
Show Gist options
  • Save cryzed/83328b876bb92d5be1ed7f3cc77a5820 to your computer and use it in GitHub Desktop.
Save cryzed/83328b876bb92d5be1ed7f3cc77a5820 to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. cryzed revised this gist May 23, 2021. 1 changed file with 2 additions and 1 deletion.
    3 changes: 2 additions & 1 deletion timedcache.py
    View file
    Original file line number Diff line number Diff line change
    @@ -19,7 +19,7 @@ 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) -> T.Any:
    def wrapped(*args: T.Hashable, **kwargs: T.Hashable) -> X:
    now = time.monotonic()
    for key, entry in tuple(cache.items()):
    if entry.expires_at >= now:
    @@ -38,3 +38,4 @@ def wrapped(*args: T.Hashable, **kwargs: T.Hashable) -> T.Any:
    return wrapped

    return _timed_cache

  2. cryzed revised this gist May 23, 2021. 1 changed file with 0 additions and 1 deletion.
    1 change: 0 additions & 1 deletion timedcache.py
    View file
    Original file line number Diff line number Diff line change
    @@ -33,7 +33,6 @@ def wrapped(*args: T.Hashable, **kwargs: T.Hashable) -> T.Any:

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

    return wrapped
  3. cryzed revised this gist May 23, 2021. 1 changed file with 1 addition and 1 deletion.
    2 changes: 1 addition & 1 deletion timedcache.py
    View file
    Original file line number Diff line number Diff line change
    @@ -33,7 +33,7 @@ def wrapped(*args: T.Hashable, **kwargs: T.Hashable) -> T.Any:

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

    return wrapped
  4. cryzed created this gist May 23, 2021.
    41 changes: 41 additions & 0 deletions timedcache.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,41 @@
    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) -> T.Any:
    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)
    cache.move_to_end(key, last=False)
    return result

    return wrapped

    return _timed_cache