cache

src.utils.cache

Shared disk-based cache backed by hashed JSON files.

Every cached value is stored as {"ts": <unix-epoch>, "body": <payload>}, optionally with an "etag" field for conditional HTTP requests.

The body may be any JSON-serialisable Python value (str, dict, list, bool, int, float, or None).

cache_path(base_dir: str, key: str, namespace: str = 'default') -> str

Return the file path for key inside base_dir/namespace/.

Source code in src/utils/cache.py

def cache_path(base_dir: str, key: str, namespace: str = "default") -> str:
    """Return the file path for *key* inside *base_dir*/*namespace*/."""
    ns_dir = os.path.join(base_dir, namespace)
    os.makedirs(ns_dir, exist_ok=True)
    hashed = hashlib.sha256(key.encode()).hexdigest()
    return os.path.join(ns_dir, hashed)

read_cache(base_dir: str, key: str, ttl: int, namespace: str = 'default') -> Any

Return cached body if fresh (younger than ttl seconds), else _MISSING.

Returns the original Python value that was stored (dict, list, str, …). Returns the module-level sentinel _MISSING on cache miss so callers can distinguish a miss from a cached None.

Source code in src/utils/cache.py

def read_cache(base_dir: str, key: str, ttl: int, namespace: str = "default") -> Any:
    """Return cached body if fresh (younger than *ttl* seconds), else ``_MISSING``.

    Returns the *original* Python value that was stored (dict, list, str, …).
    Returns the module-level sentinel ``_MISSING`` on cache miss so callers
    can distinguish a miss from a cached ``None``.
    """
    path = cache_path(base_dir, key, namespace)
    if not os.path.exists(path):
        return _MISSING
    try:
        with open(path) as f:
            entry = json.load(f)
        ts = float(entry.get("ts", 0))
        if time.time() - ts < ttl:
            return entry.get("body", _MISSING)
    except (json.JSONDecodeError, KeyError, OSError):
        pass
    return _MISSING

write_cache(base_dir: str, key: str, body: Any, namespace: str = 'default', etag: str | None = None) -> None

Write body to the cache, optionally storing an HTTP ETag.

Uses atomic write (temp file + os.replace) to avoid partial/corrupt cache files when multiple processes run concurrently.

Source code in src/utils/cache.py

def write_cache(
    base_dir: str,
    key: str,
    body: Any,
    namespace: str = "default",
    etag: str | None = None,
) -> None:
    """Write *body* to the cache, optionally storing an HTTP ETag.

    Uses atomic write (temp file + ``os.replace``) to avoid partial/corrupt
    cache files when multiple processes run concurrently.
    """
    path = cache_path(base_dir, key, namespace)
    entry: dict[str, Any] = {"ts": time.time(), "body": body}
    if etag:
        entry["etag"] = etag
    dir_name = os.path.dirname(path)
    fd, tmp_path = tempfile.mkstemp(dir=dir_name, suffix=".tmp")
    try:
        with os.fdopen(fd, "w") as f:
            json.dump(entry, f)
        os.replace(tmp_path, path)
    except BaseException:
        os.unlink(tmp_path)
        raise

read_cache_entry(base_dir: str, key: str, namespace: str = 'default') -> dict | None

Return the full cache entry dict regardless of TTL, or None.

Source code in src/utils/cache.py

def read_cache_entry(base_dir: str, key: str, namespace: str = "default") -> dict | None:
    """Return the full cache entry dict regardless of TTL, or ``None``."""
    path = cache_path(base_dir, key, namespace)
    if not os.path.exists(path):
        return None
    try:
        with open(path) as f:
            result: dict = json.load(f)
            return result
    except (json.JSONDecodeError, KeyError, OSError):
        return None

refresh_cache_ts(base_dir: str, key: str, namespace: str = 'default') -> None

Touch the timestamp of a cache entry without changing its body.

Source code in src/utils/cache.py

def refresh_cache_ts(base_dir: str, key: str, namespace: str = "default") -> None:
    """Touch the timestamp of a cache entry without changing its body."""
    path = cache_path(base_dir, key, namespace)
    try:
        with open(path) as f:
            entry = json.load(f)
        entry["ts"] = time.time()
        with open(path, "w") as f:
            json.dump(entry, f)
    except (json.JSONDecodeError, KeyError, OSError):
        pass