From bf7b82dd2bec5fc2a9766b36abc87c5a592d0404 Mon Sep 17 00:00:00 2001 From: Eddie A Tejeda <669988+eddietejeda@users.noreply.github.com> Date: Wed, 22 Jul 2026 13:56:06 -0700 Subject: [PATCH 1/2] Add CI: pytest, mypy, and flake8 on 3.11/3.12 --- .github/workflows/ci.yml | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) create mode 100644 .github/workflows/ci.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..cba42b1 --- /dev/null +++ b/.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 From 86e3ec4637cd55e9e8a51309173fc1243dfc88f0 Mon Sep 17 00:00:00 2001 From: Eddie A Tejeda <669988+eddietejeda@users.noreply.github.com> Date: Wed, 22 Jul 2026 14:00:01 -0700 Subject: [PATCH 2/2] Add @materialize decorator and MaterializedFrame - @materialize(ttl=, version=, key=, key_fn=, background=): hit returns a frame without calling the function; miss runs it, serves the result immediately, persists write-behind - MaterializedFrame: .arrow()/.df()/.to_pylist()/len(), .sql() runs server-side against the entry ("this" names the cached data) - Fail-open: registry/persist failures degrade to running uncached - EntryStore.query_table() extracted for frame SQL; README leads with the decorator --- README.md | 116 ++++++++----------- hotdata_materialized/__init__.py | 7 +- hotdata_materialized/decorator.py | 178 ++++++++++++++++++++++++++++++ hotdata_materialized/store.py | 44 +++++--- tests/test_decorator.py | 138 +++++++++++++++++++++++ 5 files changed, 397 insertions(+), 86 deletions(-) create mode 100644 hotdata_materialized/decorator.py create mode 100644 tests/test_decorator.py diff --git a/README.md b/README.md index f01e089..1c77cc3 100644 --- a/README.md +++ b/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 diff --git a/hotdata_materialized/__init__.py b/hotdata_materialized/__init__.py index 435feb1..8c9329b 100644 --- a/hotdata_materialized/__init__.py +++ b/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", diff --git a/hotdata_materialized/decorator.py b/hotdata_materialized/decorator.py new file mode 100644 index 0000000..668b7b7 --- /dev/null +++ b/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 diff --git a/hotdata_materialized/store.py b/hotdata_materialized/store.py index 78169c2..ff73c70 100644 --- a/hotdata_materialized/store.py +++ b/hotdata_materialized/store.py @@ -231,25 +231,41 @@ def read_table(self, entry: RegistryEntry) -> pa.Table: fp.short(entry.fingerprint), ) try: - response = self._clients.query.query( - QueryRequest(sql=f"SELECT * FROM {DATA_TABLE}"), - x_database_id=database_id, - auto_follow=False, - ) - if response.result_id is None: - # answered fully inline and never persisted; nothing to fetch. - # Keep the column names even for zero rows. - if not response.rows: - return pa.table({column: [] for column in response.columns}) - return pa.Table.from_pylist( - [dict(zip(response.columns, row)) for row in response.rows] - ) - return self._fetch_arrow(response.result_id, database_id) + return self._run_select(database_id, f"SELECT * FROM {DATA_TABLE}") except Exception as exc: raise StoreError( f"reading entry {fp.short(entry.fingerprint)} failed: {exc}" ) from exc + def query_table(self, entry: RegistryEntry, sql: str) -> pa.Table: + """Run SQL server-side against the entry's database, returning Arrow.""" + database_id = entry.database_id + if database_id is None: + raise StoreError( + f"entry {fp.short(entry.fingerprint)} has no database yet " + "(still building?)" + ) + try: + return self._run_select(database_id, sql) + except Exception as exc: + raise StoreError( + f"query on entry {fp.short(entry.fingerprint)} failed: {exc}" + ) from exc + + def _run_select(self, database_id: str, sql: str) -> pa.Table: + response = self._clients.query.query( + QueryRequest(sql=sql), x_database_id=database_id, auto_follow=False + ) + if response.result_id is None: + # answered fully inline and never persisted; nothing to fetch. + # Keep the column names even for zero rows. + if not response.rows: + return pa.table({column: [] for column in response.columns}) + return pa.Table.from_pylist( + [dict(zip(response.columns, row)) for row in response.rows] + ) + return self._fetch_arrow(response.result_id, database_id) + def _fetch_arrow(self, result_id: str, database_id: str) -> pa.Table: """Fetch a persisted result as Arrow, waiting out async persistence (the SDK polls status-only with limit=0) when it isn't ready yet.""" diff --git a/tests/test_decorator.py b/tests/test_decorator.py new file mode 100644 index 0000000..a7f7c3f --- /dev/null +++ b/tests/test_decorator.py @@ -0,0 +1,138 @@ +import pyarrow as pa +import pytest + +import hotdata_materialized.decorator as decorator_module +from hotdata_materialized import materialize +from hotdata_materialized.conf import Config +from hotdata_materialized.decorator import to_arrow +from hotdata_materialized.exceptions import RegistryError +from hotdata_materialized.registry import Registry +from hotdata_materialized.store import EntryStore +from tests.fakes import FakeBackend +from tests.test_registry import Clock + + +class Runtime: + def __init__(self): + self.backend = FakeBackend() + self.config = Config(api_key="hd_test", workspace_id="ws_test") + self.clock = Clock() + self.registry = Registry(self.backend, self.config, now_fn=self.clock) + self.store = EntryStore( + self.backend, self.config, self.registry, now_fn=self.clock + ) + + +@pytest.fixture +def runtime(monkeypatch): + rt = Runtime() + monkeypatch.setattr( + decorator_module, "get_runtime", lambda: (rt.registry, rt.store) + ) + return rt + + +ROWS = [{"channel": "organic", "n": 10}, {"channel": "paid", "n": 7}] + + +def make_report(calls): + @materialize(ttl=3600) + def report(day): + calls["n"] += 1 + return [dict(row, day=day) for row in ROWS] + + return report + + +def test_miss_runs_function_and_serves_result(runtime): + calls = {"n": 0} + report = make_report(calls) + frame = report("2026-07-22") + assert frame.cached is False + assert calls["n"] == 1 + assert frame.to_pylist()[0]["channel"] == "organic" + assert len(frame) == 2 + + +def test_hit_skips_the_function(runtime): + calls = {"n": 0} + report = make_report(calls) + first = report("2026-07-22") + runtime.store.flush() + second = report("2026-07-22") + assert second.cached is True + assert calls["n"] == 1 + assert second.to_pylist() == first.to_pylist() + + +def test_different_arguments_are_different_entries(runtime): + calls = {"n": 0} + report = make_report(calls) + report("2026-07-21") + runtime.store.flush() + frame = report("2026-07-22") + assert frame.cached is False + assert calls["n"] == 2 + + +def test_expired_entry_reruns_the_function(runtime): + calls = {"n": 0} + report = make_report(calls) + report("2026-07-22") + runtime.store.flush() + runtime.clock.advance(3601) + frame = report("2026-07-22") + assert frame.cached is False + assert calls["n"] == 2 + + +def test_version_busts_the_cache(runtime): + calls = {"n": 0} + + def compute(): + calls["n"] += 1 + return list(ROWS) + + report_v1 = materialize(ttl=3600, version=1)(compute) + report_v2 = materialize(ttl=3600, version=2)(compute) + report_v1() + runtime.store.flush() + frame = report_v2() + assert frame.cached is False + assert calls["n"] == 2 + + +def test_fail_open_when_registry_is_unreachable(runtime, monkeypatch): + calls = {"n": 0} + report = make_report(calls) + + def down(*args, **kwargs): + raise RegistryError("registry unavailable") + + monkeypatch.setattr(runtime.registry, "lookup", down) + frame = report("2026-07-22") + assert frame.cached is False + assert calls["n"] == 1 + assert len(frame.to_pylist()) == 2 # the caller still gets data + + +def test_sql_runs_server_side_against_the_entry(runtime): + calls = {"n": 0} + report = make_report(calls) + report("2026-07-22") + runtime.store.flush() + frame = report("2026-07-22") + assert frame.cached is True + top = frame.sql("SELECT channel FROM this ORDER BY n DESC LIMIT 1") + assert top.to_pylist() == [{"channel": "organic"}] + + +def test_to_arrow_accepts_table_and_rows(): + table = pa.table({"n": [1, 2]}) + assert to_arrow(table) is table + assert to_arrow(iter([{"n": 1}])).to_pylist() == [{"n": 1}] + + +def test_to_arrow_rejects_non_dict_rows(): + with pytest.raises(TypeError, match="iterable of dicts"): + to_arrow([1, 2, 3])