lythonic.compose.cached

Cache utilities for wrapping callables with SQLite-backed caching.

Cached: caching layer for callables via register_cached_callable.

Wraps sync or async methods that return dict or Pydantic BaseModel with SQLite-backed caching. Each cached method gets its own table with typed parameter columns (derived from the method signature) and a composite primary key.

Usage

from pathlib import Path
from lythonic.compose.cached import register_cached_callable
from lythonic.compose.namespace import Namespace

ns = Namespace()
register_cached_callable(
    ns, "myapp.downloads:fetch_prices",
    min_ttl=0.5, max_ttl=2.0, db_path=Path("cache.db"), nsref="market:fetch_prices",
)

# Sync method — served from cache when fresh
result = ns.market.fetch_prices(ticker="AAPL")

# Async method — awaited on cache miss or hard expiry
register_cached_callable(
    ns, "myapp.downloads:get_exchange_rate",
    min_ttl=0.25, max_ttl=1.0, db_path=Path("cache.db"), nsref="get_exchange_rate",
)
rate = await ns.get_exchange_rate(from_currency="USD", to_currency="EUR")

TTL Behavior

• age < min_ttl: return cached value (fresh)
• min_ttl <= age < max_ttl: probabilistic refresh — probability increases linearly from 0 to 1. On refresh failure, returns stale value.
• age >= max_ttl or cache miss: call original method. On failure, raises.

Validation

All method parameters must have types registered as simple_type in KNOWN_TYPES (primitives, date, datetime, Path). Validated at registration time via Method.validate_simple_type_args().

Pushback

When a cached method raises CacheRefreshPushback(days, namespace_prefix), all probabilistic refreshes matching the scope are suppressed for the given duration. If namespace_prefix is omitted, only the raising method is suppressed.

• During the probabilistic window with active pushback: returns stale data.
• Past max_ttl with active pushback: raises CacheRefreshSuppressed.
• Cache miss: always calls the method regardless of pushback.

Namespace

Wrapped methods are installed on a Namespace object with nested attribute access. The nsref uses colon-separated format (e.g., "market:fetch_prices" becomes ns.market.fetch_prices(...)).

CacheRefreshPushback

Bases: Exception

Raise from a cached method to suppress probabilistic refreshes. Defaults to suppressing only the raising method; set namespace_prefix to suppress a group of methods.

Source code in src/lythonic/compose/cached.py

class CacheRefreshPushback(Exception):
    """
    Raise from a cached method to suppress probabilistic refreshes.
    Defaults to suppressing only the raising method; set `namespace_prefix`
    to suppress a group of methods.
    """

    days: float
    namespace_prefix: str | None

    def __init__(self, days: float, namespace_prefix: str | None = None):
        super().__init__(f"Cache refresh pushback for {days} days")
        self.days = days
        self.namespace_prefix = namespace_prefix

CacheRefreshSuppressed

Bases: Exception

Raised when a cache entry is past max_ttl but refresh is suppressed by an active pushback.

Source code in src/lythonic/compose/cached.py

class CacheRefreshSuppressed(Exception):
    """
    Raised when a cache entry is past max_ttl but refresh is suppressed
    by an active pushback.
    """

    namespace_path: str
    suppressed_until: float

    def __init__(self, namespace_path: str, suppressed_until: float):
        super().__init__(f"Cache refresh suppressed for {namespace_path} until {suppressed_until}")
        self.namespace_path = namespace_path
        self.suppressed_until = suppressed_until

register_cached_callable(ns, gref, min_ttl, max_ttl, db_path, nsref=None)

Register a callable with cache wrapping. Handles DDL generation, wrapper building, pushback table creation, and namespace registration. Stores cache config in node.metadata["cache"].

Source code in src/lythonic/compose/cached.py

def register_cached_callable(
    ns: Namespace,
    gref: str,
    min_ttl: float,
    max_ttl: float,
    db_path: Path,
    nsref: str | None = None,
) -> NamespaceNode:
    """
    Register a callable with cache wrapping. Handles DDL generation,
    wrapper building, pushback table creation, and namespace registration.
    Stores cache config in `node.metadata["cache"]`.
    """

    gref_obj = GlobalRef(gref)
    if nsref is None:
        nsref = str(gref)
    method = Method(gref_obj)
    method.validate_simple_type_args()

    # Derive table name from nsref (convert : and . to __)
    tbl_name = table_name_from_path(nsref.replace(":", "__").replace(".", "__"))

    db_path.parent.mkdir(parents=True, exist_ok=True)

    # Create cache table and pushback table
    ddl = generate_cache_table_ddl(tbl_name, method)
    with open_sqlite_db(db_path) as conn:
        cursor = conn.cursor()
        execute_sql(cursor, ddl)
        execute_sql(
            cursor,
            "CREATE TABLE IF NOT EXISTS _pushback "
            "(namespace_prefix TEXT NOT NULL, suppressed_until REAL NOT NULL)",
        )
        conn.commit()

    min_ttl_s = min_ttl * DAYS_TO_SECONDS
    max_ttl_s = max_ttl * DAYS_TO_SECONDS

    # namespace_path for pushback matching (use nsref with : replaced by .)
    namespace_path = nsref.replace(":", ".")

    if gref_obj.is_async():
        wrapper = _build_async_wrapper(
            method, tbl_name, db_path, min_ttl_s, max_ttl_s, namespace_path
        )
    else:
        wrapper = _build_sync_wrapper(
            method, tbl_name, db_path, min_ttl_s, max_ttl_s, namespace_path
        )

    from lythonic.compose.namespace import NsCacheConfig

    cache_config = NsCacheConfig(
        nsref=nsref or str(gref_obj), gref=gref_obj, min_ttl=min_ttl, max_ttl=max_ttl
    )
    node: NamespaceNode = ns.register(gref, nsref=nsref, decorate=lambda _: wrapper)
    node.config = cache_config
    node.metadata["cache"] = {"min_ttl": min_ttl, "max_ttl": max_ttl}
    return node

generate_cache_table_ddl(table_name, method)

Generate CREATE TABLE DDL for a cache table based on the method's parameter types.

Source code in src/lythonic/compose/cached.py

def generate_cache_table_ddl(table_name: str, method: Method) -> str:
    """
    Generate CREATE TABLE DDL for a cache table based on the method's
    parameter types.
    """
    columns: list[str] = []
    param_names: list[str] = []

    for arg in method.args:
        assert arg.annotation is not None
        kt = KNOWN_TYPES.resolve_type(arg.annotation)
        db_type = kt.db_type_info.name
        columns.append(f"    {arg.name} {db_type} NOT NULL")
        param_names.append(arg.name)

    columns.append("    value_json TEXT NOT NULL")
    columns.append("    fetched_at REAL NOT NULL")

    pk = ", ".join(param_names)
    columns.append(f"    PRIMARY KEY ({pk})")

    cols_str = ",\n".join(columns)
    return f"CREATE TABLE IF NOT EXISTS {table_name} (\n{cols_str}\n)"

table_name_from_path(path)

Convert a dot-separated namespace path to a SQL table name.

Source code in src/lythonic/compose/cached.py

def table_name_from_path(path: str) -> str:
    """Convert a dot-separated namespace path to a SQL table name."""
    return path.replace(".", "__")