-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy path_cache.py
More file actions
90 lines (63 loc) · 2.01 KB
/
Copy path_cache.py
File metadata and controls
90 lines (63 loc) · 2.01 KB
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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
"""Caching module."""
from pathlib import Path
from time import time
# TOML cache file (optional)
_CACHE_FILE = Path.home() / '.secrets_cache.toml'
# Lazy-loaded cache
_cache = None
# Optional: import TOML only if available
try:
import tomllib # Python 3.11+
except ImportError:
try:
# noinspection SpellCheckingInspection
import tomli as tomllib # Python 3.10
except ImportError:
# noinspection SpellCheckingInspection
tomllib = None
try:
import tomli_w
except ImportError:
tomli_w = None
def _load_cache() -> dict:
global _cache
if _cache is None:
if tomllib and _CACHE_FILE.exists():
with _CACHE_FILE.open('rb') as f:
_cache = tomllib.load(f)
else:
_cache = {}
return _cache
def _save_cache() -> None:
if (_cache is None
or not tomli_w
or not _CACHE_FILE.parent.exists()):
return
with _CACHE_FILE.open('wb') as f:
tomli_w.dump(_cache, f) # type: ignore
def get_cached_value(service: str, name: str, now: float, ttl: int):
data = _load_cache()
service_cache = data.get(service, {})
entry = service_cache.get(name)
if entry and (now - entry['fetched_at'] < ttl):
return entry['value']
return None
def update_cache(service: str, name: str, value: str, now: float):
data = _load_cache()
final_value = {'value': value, 'fetched_at': now}
service_cache = data.get(service)
if service_cache is None:
data[service] = {name: final_value}
else:
service_cache[name] = final_value
_save_cache()
def cached_fetch(service: str, key: str, region: str, fetcher, ttl: int, force_refresh: bool):
"""Fetch from cache or call fetcher."""
now = int(time())
if not force_refresh:
if (value := get_cached_value(service, key, now, ttl)) is not None:
return value
value = fetcher(key, region)
if value is not None:
update_cache(service, key, value, now)
return value