lythonic.compose.cached

Cache utilities for wrapping callables with SQLite-backed caching.

Cached: SQLite-backed caching layer for namespace callables.

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

Declare cached callables in namespace config using NsCacheConfig, then call ns.mount(storage) to activate caching:

from pathlib import Path
from lythonic.compose.namespace import Namespace, NsCacheConfig
from lythonic.compose.engine import StorageConfig

ns = Namespace()
cfg = NsCacheConfig(
    nsref="market:fetch_prices",
    gref="myapp.downloads:fetch_prices",
    min_ttl=0.5, max_ttl=2.0,
)
ns.register("myapp.downloads:fetch_prices", nsref="market:fetch_prices", config=cfg)
ns.mount(StorageConfig(cache_db=Path("cache.db")))

result = ns.market.fetch_prices(ticker="AAPL")

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 mount 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.

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

CacheProhibitDirectCall

Bases: Exception

Raised when a cached method is called directly instead of through the cache wrapper. Call CacheProhibitDirectCall.require() from a method body to enforce this.

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

class CacheProhibitDirectCall(Exception):
    """
    Raised when a cached method is called directly instead of through
    the cache wrapper. Call `CacheProhibitDirectCall.require()` from
    a method body to enforce this.
    """

    @staticmethod
    def require() -> None:
        """Raise if not inside a cache wrapper."""
        if not _in_cache_wrapper.get():
            raise CacheProhibitDirectCall("Method must be called through cache wrapper")

require() staticmethod

Raise if not inside a cache wrapper.

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

@staticmethod
def require() -> None:
    """Raise if not inside a cache wrapper."""
    if not _in_cache_wrapper.get():
        raise CacheProhibitDirectCall("Method must be called through cache wrapper")

mount_cached_node(node, db_path)

Activate caching on a node that has NsCacheConfig. Builds the sync/async wrapper, creates DDL, and sets node._decorated. Called by Namespace.mount().

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

def mount_cached_node(node: NamespaceNode, db_path: Path) -> None:
    """
    Activate caching on a node that has `NsCacheConfig`. Builds the
    sync/async wrapper, creates DDL, and sets `node._decorated`.
    Called by `Namespace.mount()`.
    """
    from lythonic.compose.namespace import NsCacheConfig

    config = node.config
    if not isinstance(config, NsCacheConfig):
        raise TypeError(f"Expected NsCacheConfig, got {type(config).__name__}")

    method = node.method
    method.validate_simple_type_args()

    nsref = node.nsref
    tbl_name = table_name_from_path(nsref.replace(":", "__").replace(".", "__"))

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

    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 = config.min_ttl * DAYS_TO_SECONDS
    max_ttl_s = config.max_ttl * DAYS_TO_SECONDS
    namespace_path = nsref.replace(":", ".")

    gref = node.method.gref
    if gref.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
        )

    node._decorated = wrapper  # pyright: ignore[reportPrivateUsage]

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(".", "__")