"""Optional disk-based response cache for AEMet API calls.
This module provides a ``@cached`` decorator backed by a single SQLite
database file. Cached entries expire after the TTL defined in
:data:`aemetxfb.utils.CONFIG` (``cache_ttl`` seconds).
The cache is **disabled by default**; enable it via
.. code-block:: python
from aemetxfb import AEMetConfig, set_config
set_config(AEMetConfig(cache_enabled=True, cache_ttl=86400))
Only pure data-returning functions should be decorated. Functions that
perform side-effects (downloading images, writing files) must **not**
use ``@cached`` because their return values are typically file paths
that are meaningless when replayed from cache.
Typical use case — climate data that changes rarely:
.. code-block:: python
from aemetxfb.cache import cached
from aemetxfb.clim import get_normals
@cached
def my_get_normals(station: str) -> pd.DataFrame:
return get_normals(station)
"""
from __future__ import annotations
import hashlib
import pickle
import sqlite3
import time
from functools import wraps
from pathlib import Path
from typing import Any
from aemetxfb import utils as _utils
# ---------------------------------------------------------------------------
# SQLite helpers
# ---------------------------------------------------------------------------
_DB_SCHEMA = """
CREATE TABLE IF NOT EXISTS cache (
key TEXT PRIMARY KEY,
value BLOB NOT NULL,
created REAL NOT NULL
);
"""
def _get_db_path() -> Path:
"""Return the path to the SQLite cache database file."""
cache_dir = Path(_utils.CONFIG.cache_dir)
cache_dir.mkdir(parents=True, exist_ok=True)
return cache_dir / "cache.db"
def _get_connection() -> sqlite3.Connection:
"""Open (and initialise) the SQLite connection."""
conn = sqlite3.connect(str(_get_db_path()))
conn.execute(_DB_SCHEMA)
conn.commit()
return conn
def _make_key(func_name: str, args: tuple, kwargs: tuple) -> str:
"""Create a deterministic cache key from function name and arguments.
The key is the SHA-256 hex digest of a pickled tuple containing the
function name, positional arguments, and sorted keyword arguments.
"""
raw = (func_name, args, kwargs)
digest = hashlib.sha256(pickle.dumps(raw)).hexdigest()
return digest
# ---------------------------------------------------------------------------
# Low-level cache API
# ---------------------------------------------------------------------------
[docs]
def cache_get(key: str) -> Any | None:
"""Retrieve a value from the cache by key.
Returns ``None`` if the key is missing or has expired.
Examples
--------
>>> cache_get("nonexistent_key") is None # cache disabled by default
True
"""
if not _utils.CONFIG.cache_enabled:
return None
conn = _get_connection()
try:
row = conn.execute(
"SELECT value, created FROM cache WHERE key = ?", (key,)
).fetchone()
if row is None:
return None
value, created = row
age = time.time() - created
if age > _utils.CONFIG.cache_ttl:
conn.execute("DELETE FROM cache WHERE key = ?", (key,))
conn.commit()
return None
return pickle.loads(value)
finally:
conn.close()
[docs]
def cache_set(key: str, value: Any) -> None:
"""Store a value in the cache.
If the key already exists it is overwritten.
"""
if not _utils.CONFIG.cache_enabled:
return
conn = _get_connection()
try:
conn.execute(
"INSERT OR REPLACE INTO cache (key, value, created) VALUES (?, ?, ?)",
(key, pickle.dumps(value), time.time()),
)
conn.commit()
finally:
conn.close()
[docs]
def clear_cache() -> None:
"""Delete all entries from the cache database.
The database file itself is kept so that subsequent inserts do not
need to re-create the schema.
Examples
--------
>>> clear_cache() # safe to call even when cache is disabled
"""
conn = _get_connection()
try:
conn.execute("DELETE FROM cache")
conn.commit()
finally:
conn.close()
[docs]
def get_cache_stats() -> dict[str, int]:
"""Return basic statistics about the current cache.
Returns
-------
dict
``{"size": ..., "entries": ...}`` where *size* is the file size
in bytes and *entries* is the number of rows in the cache table.
If the database does not exist yet both values are ``0``.
Examples
--------
>>> stats = get_cache_stats()
>>> "size" in stats and "entries" in stats
True
"""
db_path = _get_db_path()
if not db_path.exists():
return {"size": 0, "entries": 0}
size = db_path.stat().st_size
conn = _get_connection()
try:
count = conn.execute("SELECT COUNT(*) FROM cache").fetchone()[0]
finally:
conn.close()
return {"size": size, "entries": count}
# ---------------------------------------------------------------------------
# Decorator
# ---------------------------------------------------------------------------
[docs]
def cached(func: Any) -> Any:
"""Decorator that caches the return value of a function.
The cache key is derived from the function name, positional arguments,
and keyword arguments. Entries expire after ``CONFIG.cache_ttl``
seconds. If ``CONFIG.cache_enabled`` is ``False`` the decorator is a
no-op and the underlying function is called every time.
Parameters that are not hashable (e.g. mutable lists) will cause a
``TypeError`` when pickling the cache key.
Examples
--------
>>> from aemetxfb import AEMetConfig, set_config # doctest: +SKIP
>>> from aemetxfb.cache import cached # doctest: +SKIP
>>> set_config(AEMetConfig(cache_enabled=True, cache_ttl=3600)) # doctest: +SKIP
>>> @cached # doctest: +SKIP
... def fetch_normals(station: str): # doctest: +SKIP
... return get_normals(station) # doctest: +SKIP
"""
@wraps(func)
def wrapper(*args, **kwargs):
sorted_kwargs = tuple(sorted(kwargs.items()))
key = _make_key(func.__qualname__, args, sorted_kwargs)
cached_value = cache_get(key)
if cached_value is not None:
return cached_value
result = func(*args, **kwargs)
cache_set(key, result)
return result
return wrapper