|
26 | 26 | # OR OTHER DEALINGS IN THE SOFTWARE. |
27 | 27 | # |
28 | 28 |
|
29 | | -# stdlib |
30 | | -import inspect |
31 | | -import json |
32 | | -import shutil |
33 | | -import warnings |
34 | | -from functools import wraps |
35 | | -from typing import Any, Callable, Dict, Iterable, Optional |
36 | | - |
37 | 29 | # 3rd party |
38 | | -from appdirs import user_cache_dir |
39 | | -from domdf_python_tools.paths import PathPlus |
40 | | -from domdf_python_tools.utils import posargs2kwargs |
41 | | - |
42 | | -__all__ = ["Cache"] |
43 | | - |
44 | | - |
45 | | -class Cache: |
46 | | - |
47 | | - def __init__(self, app_name: str): |
48 | | - self.app_name: str = str(app_name) |
49 | | - self.cache_dir = PathPlus(user_cache_dir(f"{self.app_name}_cache")) |
50 | | - self.cache_dir.maybe_make(parents=True) |
51 | | - |
52 | | - # Mapping of function names to their caches |
53 | | - self.caches: Dict[str, Dict[str, Any]] = {} |
54 | | - |
55 | | - def clear(self, func: Optional[Callable] = None) -> bool: |
56 | | - """ |
57 | | - Clear the cache. |
58 | | -
|
59 | | - :param func: Optional function to clear the cache for. |
60 | | - By default, the whole cache is cleared. |
61 | | - :no-default func: |
62 | | -
|
63 | | - :returns: True to indicate success. False otherwise. |
64 | | - """ |
65 | | - |
66 | | - try: |
67 | | - if func is None: |
68 | | - shutil.rmtree(self.cache_dir) |
69 | | - self.cache_dir.maybe_make() |
70 | | - for function in self.caches: |
71 | | - del self.caches[function] |
72 | | - self.caches[function] = {} |
73 | | - else: |
74 | | - function_name = func.__name__ |
75 | | - cache_file = self.cache_dir / f"{function_name}.json" |
76 | | - if cache_file.is_file(): |
77 | | - cache_file.unlink() |
78 | | - |
79 | | - if function_name in self.caches: |
80 | | - del self.caches[function_name] |
81 | | - self.caches[function_name] = {} |
82 | | - |
83 | | - return True |
84 | | - |
85 | | - except Exception as e: # pragma: no cover |
86 | | - warnings.warn(f"Could not remove cache. The error was: {e}") |
87 | | - return False |
88 | | - |
89 | | - def load_cache(self, func: Callable) -> None: |
90 | | - """ |
91 | | - Loads the cache for the given function |
92 | | -
|
93 | | - :param func: |
94 | | - """ |
95 | | - |
96 | | - cache_file: PathPlus = self.cache_dir / f"{func.__name__}.json" |
97 | | - |
98 | | - if cache_file.is_file(): |
99 | | - cache = json.loads(cache_file.read_text()) |
100 | | - else: |
101 | | - cache = {} |
102 | | - |
103 | | - self.caches[func.__name__] = cache |
104 | | - return cache |
105 | | - |
106 | | - def __call__(self, func: Callable): |
107 | | - """ |
108 | | - Decorator to cache the return values of a function based on its inputs. |
109 | | -
|
110 | | - :param func: |
111 | | - """ |
112 | | - |
113 | | - function_name = func.__name__ |
114 | | - posargs: Iterable[str] = inspect.getfullargspec(func).args |
115 | | - cache_file: PathPlus = self.cache_dir / f"{function_name}.json" |
116 | | - self.load_cache(func) |
117 | | - |
118 | | - @wraps(func) |
119 | | - def wrapper(*args, **kwargs): |
120 | | - kwargs: Dict[str, Any] = posargs2kwargs(args, posargs, kwargs) # type: ignore |
121 | | - key: str = json.dumps(kwargs) |
122 | | - response: Any |
123 | | - |
124 | | - cache = self.caches[function_name] |
125 | | - if key in cache: |
126 | | - # Return cached response |
127 | | - response = json.loads(cache[key]) |
128 | | - else: |
129 | | - response = func(**kwargs) |
130 | | - |
131 | | - if response is not None: |
132 | | - # Don't cache None values. |
133 | | - cache[key] = json.dumps(response) |
134 | | - |
135 | | - cache_file.write_text(json.dumps(cache)) |
136 | | - |
137 | | - return response |
138 | | - |
139 | | - return wrapper |
| 30 | +from apeye.cache import Cache |
140 | 31 |
|
| 32 | +__all__ = ["cache"] |
141 | 33 |
|
142 | 34 | cache = Cache("seed_intersphinx_mapping") |
0 commit comments