CI workflow + @materialize decorator

.github/workflows/ci.yml

@@ -0,0 +1,30 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        # tests use tomllib (3.11+); the library itself supports >=3.9
+        python-version: ["3.11", "3.12"]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install
+        run: |
+          python -m pip install --upgrade pip
+          pip install -e '.[test]' setuptools wheel
+      - name: flake8
+        run: flake8 hotdata_materialized/ tests/ demo/
+      - name: mypy
+        run: mypy
+      - name: pytest
+        run: pytest -q

README.md

@@ -44,95 +44,70 @@ HOTDATA_MATERIALIZED = {
 
 ## Using it in a Django view
 
-The `@materialize` decorator is coming (see [DESIGN.md](DESIGN.md)); today the
-primitives compose in a few lines. The pattern: fingerprint the queryset,
-check the registry, read the entry on a hit, run-and-materialize on a miss.
+Wrap the expensive part in `@materialize`; call it from your view:
 
 ```python
-# views.py
-import pyarrow as pa
+# reports.py
 from django.db.models import Count, Sum
-from django.http import JsonResponse
-
-from hotdata_materialized import Config, EntryStore, Registry, get_clients
-from hotdata_materialized.fingerprint import fingerprint_queryset
-from hotdata_materialized.registry import utcnow_iso
+from hotdata_materialized import materialize
 
 from .models import Order
 
 
-def _cache():
-    config = Config.from_django()
-    clients = get_clients(config)
-    registry = Registry(clients, config)
-    return registry, EntryStore(clients, config, registry)
-
-
-def revenue_by_region(request):
-    queryset = (
+@materialize(ttl=3600)
+def revenue_by_region():
+    return (
         Order.objects
         .filter(status="complete")
         .values("region")
         .annotate(orders=Count("id"), revenue=Sum("total"))
         .order_by("-revenue")
     )
-
-    registry, store = _cache()
-    fingerprint = fingerprint_queryset(queryset)
-
-    entry = registry.lookup(fingerprint)
-    if entry and entry.status == "ready" and not entry.is_expired(utcnow_iso()):
-        # HIT: your database is never touched. Small results decode locally
-        # from the registry row; large ones stream back as Arrow.
-        table = store.read_table(entry)
-        return JsonResponse({"rows": table.to_pylist(), "cached": True})
-
-    # MISS: the query runs on your database, exactly as it would uncached.
-    rows = list(queryset)
-    store.materialize(
-        fingerprint,
-        pa.Table.from_pylist(rows),
-        key="revenue-by-region",   # human-readable label in the registry
-        ttl=3600,                  # None = never expires
-    )
-    # materialize() returned immediately — the parquet upload and registry
-    # write are happening in a background thread while you respond.
-    return JsonResponse({"rows": rows, "cached": False})
 ```
 
-Notes on the moving parts:
+```python
+# views.py
+from django.http import JsonResponse
+from .reports import revenue_by_region
 
-- **`fingerprint_queryset(qs)`** hashes the compiled `(sql, params)` — no
-  cache keys to invent, and two different querysets can't collide. For
-  caching a plain function's result use `fingerprint_call(func, args, kwargs,
-  version=...)`; bump `version=` to bust the cache on a code change.
-- **`materialize()` is write-behind by default**: it returns a pending entry
-  immediately and persists in the background. Pass `background=False` to
-  block until the entry is ready (e.g. in a management command), and use
-  `store.flush()` in tests or scripts to wait for in-flight persists — it
-  returns any exceptions they raised.
-- **Dataframes:** `store.read_table(entry)` returns a `pyarrow.Table`;
-  `.to_pandas()` / `polars.from_arrow(...)` from there.
 
-## Working with a cached entry beyond fetching it
+def revenue_view(request):
+    frame = revenue_by_region()
+    return JsonResponse({"rows": frame.to_pylist(), "cached": frame.cached})
+```
 
-Every entry is its own Hotdata database with the data at `main.data`, so you
-can push transformations to Hotdata instead of pulling rows back:
+On a **hit** the function never runs and your database is never touched. On a
+**miss** the function runs exactly as it would uncached, the caller gets the
+result immediately, and the persist happens in a background thread. Either
+way you get a `MaterializedFrame`:
 
-```python
-from hotdata.models.query_request import QueryRequest
+- `frame.arrow()` — the data as a `pyarrow.Table` (`frame.df()` for pandas,
+  `frame.to_pylist()` for dicts, `len(frame)` for the row count)
+- `frame.sql("SELECT region, revenue FROM this ORDER BY revenue DESC LIMIT 5")`
+  — SQL runs server-side against the cached entry; `this` names the data
+- `frame.cached` — which path served it; `frame.entry` — the registry record
 
-clients = get_clients(Config.from_django())
-top5 = clients.query.query(
-    QueryRequest(sql="SELECT region, revenue FROM data ORDER BY revenue DESC LIMIT 5"),
-    x_database_id=entry.database_id,
-)
-```
+The wrapped function can return an iterable of dicts (a `.values()` queryset),
+a `pyarrow.Table`, or a pandas DataFrame. Decorator knobs: `ttl=` (seconds,
+`None` = never expires), `version=` (bump to bust the cache on code changes),
+`key=` (human-readable label), `key_fn=` (stable identity for arguments that
+aren't JSON-serializable), `background=False` (block until persisted).
+Fail-open by design: if Hotdata is unreachable, the function runs and its
+result is served uncached — the cache degrades to "no cache," never to
+"no page."
+
+## Evicting entries
 
-Refresh or drop an entry explicitly:
+The decorator composes public primitives (`fingerprint_call` /
+`fingerprint_queryset`, `Registry`, `EntryStore`) that you can use directly —
+for example to drop an entry explicitly:
 
 ```python
-store.evict(fingerprint)   # delete the registry row and the entry database
+from hotdata_materialized import fingerprint_call
+from hotdata_materialized.decorator import get_runtime
+
+registry, store = get_runtime()
+store.evict(fingerprint_call(revenue_by_region))
 ```
 
 Expired entries stop being served immediately (the `is_expired` check) and
@@ -154,10 +129,11 @@ trade-offs.
 
 ## Status / roadmap
 
-First draft. Implemented: fingerprinting, the remote registry, entry store
-with write-behind persists, Arrow-native reads, and the TPC-H demo. Next: the
-`@materialize` decorator and `MaterializedFrame` handle, stale-while-
-revalidate refresh, the sweep command, and vector/BM25 index declarations.
+First draft. Implemented: the `@materialize` decorator and
+`MaterializedFrame` (first cut), fingerprinting, the remote registry, entry
+store with write-behind persists, Arrow-native reads, and the TPC-H demo.
+Next: stale-while-revalidate refresh, the sweep command, a chainable queryset
+facade on the frame, and vector/BM25 index declarations.
 
 ## Demo
 

hotdata_materialized/__init__.py

@@ -1,11 +1,12 @@
 """hotdata-materialized: offload expensive Django query results to Hotdata.
 
-Step-1 surface: fingerprinting, the remote registry, and the entry store.
-The @materialize decorator and MaterializedFrame build on these (step 2).
+Public surface: the @materialize decorator and MaterializedFrame handle,
+built on fingerprinting, the remote registry, and the entry store.
 """
 
 from .conf import Config
 from .client import HotdataClients, get_clients, reset_clients
+from .decorator import MaterializedFrame, materialize
 from .exceptions import (
     ConfigurationError,
     FingerprintError,
@@ -20,6 +21,8 @@
 __version__ = "0.1.0"
 
 __all__ = [
+    "materialize",
+    "MaterializedFrame",
     "Config",
     "HotdataClients",
     "get_clients",

hotdata_materialized/decorator.py

@@ -0,0 +1,178 @@
+"""The @materialize decorator and the MaterializedFrame handle.
+
+Wrap an expensive function with @materialize: a hit returns a frame backed by
+Hotdata without calling the function; a miss calls it, returns a frame over
+the computed result immediately, and persists write-behind. The wrapped
+function must return something table-shaped: a pyarrow.Table, a pandas
+DataFrame, or an iterable of dicts (e.g. a Django .values() queryset).
+
+Fail-open: if the hit check or the persist raises a MaterializedError, the
+function runs and its result is served uncached — the cache can degrade to
+"no cache," never to "no page."
+"""
+
+from __future__ import annotations
+
+import functools
+import logging
+import re
+import threading
+from typing import Any, Callable, Dict, List, Optional, Tuple
+
+import pyarrow as pa
+
+from .client import get_clients
+from .conf import Config
+from .exceptions import MaterializedError
+from .fingerprint import fingerprint_call
+from .registry import STATUS_READY, Registry, RegistryEntry
+from .store import DATA_TABLE, EntryStore
+
+logger = logging.getLogger(__name__)
+
+_THIS = re.compile(r"\bthis\b")
+
+
+class MaterializedFrame:
+    """Handle over a materialized entry. One interface, two backings: a fresh
+    miss wraps the locally computed table; a hit reads from Hotdata lazily."""
+
+    def __init__(
+        self,
+        store: EntryStore,
+        entry: RegistryEntry,
+        table: Optional[pa.Table] = None,
+        *,
+        cached: bool,
+    ):
+        self._store = store
+        self.entry = entry
+        self._table = table
+        self.cached = cached
+
+    def arrow(self) -> pa.Table:
+        if self._table is None:
+            self._table = self._store.read_table(self.entry)
+        return self._table
+
+    def to_pylist(self) -> List[Dict[str, Any]]:
+        return self.arrow().to_pylist()
+
+    def df(self) -> Any:
+        """The data as a pandas.DataFrame (requires pandas)."""
+        return self.arrow().to_pandas()
+
+    def sql(self, sql: str) -> pa.Table:
+        """Run SQL server-side against this entry's database; the identifier
+        `this` names the cached data (e.g. "SELECT x FROM this LIMIT 5")."""
+        return self._store.query_table(self.entry, _THIS.sub(DATA_TABLE, sql))
+
+    def __len__(self) -> int:
+        return self.arrow().num_rows
+
+
+_runtime_lock = threading.Lock()
+_runtime: Optional[Tuple[Registry, EntryStore]] = None
+
+
+def get_runtime() -> Tuple[Registry, EntryStore]:
+    """The process-wide (Registry, EntryStore) pair, built from Django
+    settings on first use."""
+    global _runtime
+    with _runtime_lock:
+        if _runtime is None:
+            config = Config.from_django()
+            clients = get_clients(config)
+            registry = Registry(clients, config)
+            _runtime = (registry, EntryStore(clients, config, registry))
+        return _runtime
+
+
+def reset_runtime() -> None:
+    global _runtime
+    with _runtime_lock:
+        _runtime = None
+
+
+def to_arrow(result: Any) -> pa.Table:
+    """Normalize a captured result into a pyarrow.Table."""
+    if isinstance(result, pa.Table):
+        return result
+    if type(result).__module__.split(".")[0] == "pandas":
+        return pa.Table.from_pandas(result)
+    rows = list(result)
+    if rows and not isinstance(rows[0], dict):
+        raise TypeError(
+            "materialize expects a pyarrow.Table, a pandas.DataFrame, or an "
+            "iterable of dicts (e.g. a .values() queryset); got an iterable "
+            f"of {type(rows[0]).__name__}"
+        )
+    return pa.Table.from_pylist(rows)
+
+
+def materialize(
+    key: Optional[str] = None,
+    *,
+    ttl: Optional[int] = 3600,
+    version: int = 0,
+    background: Optional[bool] = None,
+    key_fn: Optional[Callable[..., Any]] = None,
+) -> Callable[[Callable[..., Any]], Callable[..., MaterializedFrame]]:
+    """Materialize a function's result into Hotdata.
+
+        @materialize(ttl=3600)
+        def revenue_by_region():
+            return Order.objects.values("region").annotate(revenue=Sum("total"))
+
+    The call returns a MaterializedFrame either way; `.cached` says which path
+    served it. `version=` busts the cache on code changes; `key_fn=` maps
+    non-JSON-serializable arguments to a stable identity.
+    """
+
+    def decorate(func: Callable[..., Any]) -> Callable[..., MaterializedFrame]:
+        @functools.wraps(func)
+        def wrapper(*args: Any, **kwargs: Any) -> MaterializedFrame:
+            registry, store = get_runtime()
+            fingerprint = fingerprint_call(
+                func, args, kwargs, version=version, key_fn=key_fn
+            )
+            entry = None
+            try:
+                entry = registry.lookup(fingerprint)
+            except MaterializedError:
+                logger.warning(
+                    "hit check for %s failed; running uncached",
+                    func.__qualname__,
+                    exc_info=True,
+                )
+            if (
+                entry is not None
+                and entry.status == STATUS_READY
+                and not entry.is_expired(store._now())
+            ):
+                return MaterializedFrame(store, entry, cached=True)
+
+            result = func(*args, **kwargs)
+            table = to_arrow(result)
+            label = key or func.__qualname__
+            try:
+                pending = store.materialize(
+                    fingerprint,
+                    table,
+                    key=label,
+                    ttl=ttl,
+                    version=version,
+                    background=background,
+                )
+            except MaterializedError:
+                logger.warning(
+                    "persist of %s failed; serving the result uncached",
+                    label,
+                    exc_info=True,
+                )
+                pending = RegistryEntry(fingerprint=fingerprint, key=label)
+            return MaterializedFrame(store, pending, table=table, cached=False)
+
+        return wrapper
+
+    return decorate