File: cache.py

package info (click to toggle)
python-apischema 0.18.3-3
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 1,636 kB
  • sloc: python: 15,281; makefile: 3; sh: 2
file content (52 lines) | stat: -rw-r--r-- 1,153 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
__all__ = ["cache", "reset", "set_size"]
import sys
from functools import lru_cache
from typing import Callable, Iterator, MutableMapping, TypeVar, cast

_cached: list = []

Func = TypeVar("Func", bound=Callable)


def cache(func: Func) -> Func:
    cached = cast(Func, lru_cache()(func))
    _cached.append(cached)
    return cached


def reset():
    for cached in _cached:
        cached.cache_clear()


def set_size(size: int):
    for cached in _cached:
        wrapped = cached.__wrapped__
        setattr(
            sys.modules[wrapped.__module__], wrapped.__name__, lru_cache(size)(wrapped)
        )


K = TypeVar("K")
V = TypeVar("V")


class CacheAwareDict(MutableMapping[K, V]):
    def __init__(self, wrapped: MutableMapping[K, V]):
        self.wrapped = wrapped

    def __getitem__(self, key: K) -> V:
        return self.wrapped[key]

    def __setitem__(self, key: K, value: V):
        self.wrapped[key] = value
        reset()

    def __delitem__(self, key: K):
        del self.wrapped[key]

    def __len__(self) -> int:
        return len(self.wrapped)

    def __iter__(self) -> Iterator[K]:
        return iter(self.wrapped)