From 7989f142a12cd0c628347e496dda6cac35a6b7ee Mon Sep 17 00:00:00 2001 From: Emi Griffith Date: Sat, 11 Jul 2026 12:49:15 -0700 Subject: [PATCH] Single-source cache identity; shared fetch preamble and cache flow (#43) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The derived store's key/token formats existed in three places — each endpoint, api_cell's slice assembly, and migrate.py — where any drift would silently split the cache (endpoints missing rows the bundle wrote, migrate materializing rows nobody reads). They are now defined once in views.py (grade_key/calendar_key/day_key/forecast_key, history_token/ recent_token/day_token) and consumed everywhere, with a pinning test so a format change is always deliberate. The four data endpoints shared two copy-pasted sequences, now helpers: - _fetch_history: history (+ optional recent bundle) fetch with upstream failures mapped to clean HTTP errors and an empty record to 404. - _cached_response: If-None-Match 304 / token-valid store replay / build + persist + serve, with calendar's dont-persist-placeless rule as an explicit flag. Each endpoint is now its audit run + identity + a build callback (~10 lines); api_day's hourly-token special case moved into day_token. New tests: identity format pins, rate-limit 503 parametrized across all five data routes, prefetch=1 never touching upstream (cold 204, warm history-only slices), and calendar's placeless-payload retry behavior. --- app.py | 224 ++++++++++++++++++-------------------------- migrate.py | 6 +- tests/test_api.py | 34 ++++++- tests/test_views.py | 26 +++++ views.py | 45 +++++++++ 5 files changed, 198 insertions(+), 137 deletions(-) diff --git a/app.py b/app.py index a8c40d4..a389ead 100644 --- a/app.py +++ b/app.py @@ -5,7 +5,6 @@ import functools import hashlib import json import os -import time import pandas as pd from fastapi import APIRouter, FastAPI, HTTPException, Query, Request, Response @@ -121,6 +120,51 @@ def _json_response(body: bytes, etag: str | None = None) -> Response: return Response(content=body, media_type="application/json", headers=headers) +def _fetch_history(run, cell, recent_too=False, recent_phase="recent"): + """The shared fetch preamble for every data route: the archive record (and + optionally the hourly recent/forecast bundle) with upstream failures mapped + to clean HTTP errors and an empty record to a 404.""" + try: + with run.phase("history"): + history, cache_meta = climate.get_history(cell) + recent = None + if recent_too: + with run.phase(recent_phase): + recent = climate.get_recent_forecast(cell) + except Exception as e: # noqa: BLE001 + raise _weather_fetch_error(e) + if history.empty: + raise HTTPException(status_code=404, detail="No historical data for this cell.") + return history, cache_meta, recent + + +def _cached_response(request, run, kind, cell, key, token, build, cache_placeless=True): + """The shared derived-store flow behind every per-view endpoint: + If-None-Match -> empty 304; a token-valid store row -> replay its exact + bytes; otherwise resolve the place label, build the payload, persist, serve. + + ``build(place) -> payload`` does any endpoint-specific audit tagging itself. + ``cache_placeless=False`` skips persisting a payload whose place label + failed to resolve (a transient reverse-geocode miss) — otherwise the + bare-coordinates fallback would stick for the life of the token.""" + cid = cell["id"] + etag = _etag_for(kind, cid, key, token) + if _not_modified(request, etag): + run.set(run_type="cache", not_modified=True) + return Response(status_code=304, headers={"ETag": etag}) + body = store.get_payload(kind, cid, key, token) + if body is not None: + run.set(run_type="cache", history_source="store") + return _json_response(body, etag) + with run.phase("reverse_geocode"): + place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) + payload = build(place) + return _json_response( + store.put_payload(kind, cid, key, token, payload, + cache=cache_placeless or place is not None), etag) + + + # --- endpoints ---------------------------------------------------------------- def api_geocode(q: str = Query(..., min_length=1)): @@ -255,32 +299,12 @@ def api_grade( recent_days=days, cell_id=cell["id"], ) as run: - try: - with run.phase("history"): - history, cache_meta = climate.get_history(cell) - with run.phase("recent"): - recent = climate.get_recent_forecast(cell) - except Exception as e: # noqa: BLE001 - raise _weather_fetch_error(e) - - if history.empty: - raise HTTPException(status_code=404, detail="No historical data for this cell.") - - key = f"{target.date().isoformat()}:{days}:{after}" - token = f"{views.PAYLOAD_VER}:{views.hist_end(history)}:{climate.recent_stamp(cell['id'])}" - etag = _etag_for("grade", cell["id"], key, token) - if _not_modified(request, etag): - run.set(run_type="cache", not_modified=True) - return Response(status_code=304, headers={"ETag": etag}) - body = store.get_payload("grade", cell["id"], key, token) - if body is not None: - run.set(run_type="cache", history_source="store") - return _json_response(body, etag) - - with run.phase("reverse_geocode"): - place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) - payload = views.build_grade(cell, target, days, history, recent, cache_meta, place, run, after=after) - return _json_response(store.put_payload("grade", cell["id"], key, token, payload), etag) + history, cache_meta, recent = _fetch_history(run, cell, recent_too=True) + return _cached_response( + request, run, "grade", cell, + views.grade_key(target, days, after), views.recent_token(history, cell["id"]), + lambda place: views.build_grade(cell, target, days, history, recent, + cache_meta, place, run, after=after)) def api_calendar( @@ -310,39 +334,22 @@ def api_calendar( recent_days=months * 31, # approximate span graded cell_id=cell["id"], ) as run: - try: - with run.phase("history"): - history, cache_meta = climate.get_history(cell) - except Exception as e: # noqa: BLE001 - raise _weather_fetch_error(e) - if history.empty: - raise HTTPException(status_code=404, detail="No historical data for this cell.") - + history, cache_meta, _ = _fetch_history(run, cell) start_ts, end_ts = views.cal_span(history, start, end, months) - key = f"{start_ts.date().isoformat()}:{end_ts.date().isoformat()}:{months}" - token = f"{views.PAYLOAD_VER}:{views.hist_end(history)}" - etag = _etag_for("calendar", cell["id"], key, token) - if _not_modified(request, etag): - run.set(run_type="cache", not_modified=True) - return Response(status_code=304, headers={"ETag": etag}) - body = store.get_payload("calendar", cell["id"], key, token) - if body is not None: - run.set(run_type="cache", history_source="store") - return _json_response(body, etag) - with run.phase("reverse_geocode"): - place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) - payload = views.build_calendar(cell, history, start_ts, end_ts, months, place, run) - full = not cache_meta.get("cached", False) - run.set(run_type="full" if full else "partial", - history_source="fetch" if full else "cache") - # Don't persist a payload whose place failed to resolve (a transient reverse- - # geocode miss) — otherwise the bare-coordinates fallback would stick for the - # life of the token. Compare loads several cells at once, so this is where a - # miss is most likely; leaving it uncached lets the next request retry. - return _json_response( - store.put_payload("calendar", cell["id"], key, token, payload, - cache=place is not None), etag) + def build(place): + payload = views.build_calendar(cell, history, start_ts, end_ts, months, place, run) + full = not cache_meta.get("cached", False) + run.set(run_type="full" if full else "partial", + history_source="fetch" if full else "cache") + return payload + + # Compare loads several cells at once, so a transient reverse-geocode miss + # is most likely here; cache_placeless=False keeps that miss un-persisted. + return _cached_response(request, run, "calendar", cell, + views.calendar_key(start_ts, end_ts, months), + views.history_token(history), build, + cache_placeless=False) def api_day( @@ -367,38 +374,21 @@ def api_day( lon=round(lon, 4), cell_id=cell["id"], ) as run: - try: - with run.phase("history"): - history, cache_meta = climate.get_history(cell) - except Exception as e: # noqa: BLE001 - raise _weather_fetch_error(e) - if history.empty: - raise HTTPException(status_code=404, detail="No historical data for this cell.") - + history, cache_meta, _ = _fetch_history(run, cell) last = pd.Timestamp(history["date"].max()).normalize() target = pd.Timestamp(date).normalize() if date else last run.set(target_date=target.date().isoformat()) - key = target.date().isoformat() - token = f"{views.PAYLOAD_VER}:{views.hist_end(history)}" - if target > last: - token += f":h{int(time.time() // 3600)}" # obs from the hourly recent bundle - etag = _etag_for("day", cell["id"], key, token) - if _not_modified(request, etag): - run.set(run_type="cache", not_modified=True) - return Response(status_code=304, headers={"ETag": etag}) - body = store.get_payload("day", cell["id"], key, token) - if body is not None: - run.set(run_type="cache", history_source="store") - return _json_response(body, etag) + def build(place): + payload = views.build_day(cell, history, target, place, run) + full = not cache_meta.get("cached", False) + run.set(run_type="full" if full else "partial", + history_source="fetch" if full else "cache") + return payload - with run.phase("reverse_geocode"): - place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) - payload = views.build_day(cell, history, target, place, run) - full = not cache_meta.get("cached", False) - run.set(run_type="full" if full else "partial", - history_source="fetch" if full else "cache") - return _json_response(store.put_payload("day", cell["id"], key, token, payload), etag) + return _cached_response(request, run, "day", cell, + views.day_key(target), views.day_token(history, target), + build) def api_forecast( @@ -419,34 +409,17 @@ def api_forecast( with audit.RunAudit(endpoint="forecast", lat=round(lat, 4), lon=round(lon, 4), recent_days=days, cell_id=cell["id"]) as run: - try: - with run.phase("history"): - history, _ = climate.get_history(cell) - with run.phase("forecast"): - fc = climate.get_recent_forecast(cell) - except Exception as e: # noqa: BLE001 - raise _weather_fetch_error(e) - - if history.empty: - raise HTTPException(status_code=404, detail="No historical data for this cell.") - + history, _, fc = _fetch_history(run, cell, recent_too=True, recent_phase="forecast") today = pd.Timestamp(datetime.date.today()) - key = f"{today.date().isoformat()}:{days}" - token = f"{views.PAYLOAD_VER}:{views.hist_end(history)}:{climate.recent_stamp(cell['id'])}" - etag = _etag_for("forecast", cell["id"], key, token) - if _not_modified(request, etag): - run.set(run_type="cache", not_modified=True) - return Response(status_code=304, headers={"ETag": etag}) - body = store.get_payload("forecast", cell["id"], key, token) - if body is not None: - run.set(run_type="cache", history_source="store") - return _json_response(body, etag) - with run.phase("reverse_geocode"): - place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) - payload = views.build_forecast(cell, days, history, fc, today, place, run) - run.set(run_type="partial") - return _json_response(store.put_payload("forecast", cell["id"], key, token, payload), etag) + def build(place): + payload = views.build_forecast(cell, days, history, fc, today, place, run) + run.set(run_type="partial") + return payload + + return _cached_response(request, run, "forecast", cell, + views.forecast_key(today, days), + views.recent_token(history, cell["id"]), build) def api_cell( @@ -485,18 +458,9 @@ def api_cell( return Response(status_code=204) cache_meta = {"cached": True} else: - try: - with run.phase("history"): - history, cache_meta = climate.get_history(cell) - with run.phase("recent"): - recent = climate.get_recent_forecast(cell) - except Exception as e: # noqa: BLE001 - raise _weather_fetch_error(e) - if history.empty: - raise HTTPException(status_code=404, detail="No historical data for this cell.") + history, cache_meta, recent = _fetch_history(run, cell, recent_too=True) cid = cell["id"] - hend = views.hist_end(history) last = pd.Timestamp(history["date"].max()).normalize() with run.phase("reverse_geocode"): @@ -513,34 +477,30 @@ def api_cell( return {"etag": etag, "data": data} slices = {} - hist_token = f"{views.PAYLOAD_VER}:{hend}" + hist_token = views.history_token(history) # Calendar: the default last-24-months span — what the calendar view (and # the cross-view prefetch) requests first. start_ts, end_ts = views.cal_span(history, None, None, 24) - cal_key = f"{start_ts.date().isoformat()}:{end_ts.date().isoformat()}:24" slices["calendar"] = slice_for( - "calendar", cal_key, hist_token, + "calendar", views.calendar_key(start_ts, end_ts, 24), hist_token, lambda: views.build_calendar(cell, history, start_ts, end_ts, 24, place, run)) if prefetch: # Latest archived day: its observation comes from history alone, so # it's buildable without the (skipped) recent bundle. slices["day"] = slice_for( - "day", last.date().isoformat(), hist_token, + "day", views.day_key(last), hist_token, lambda: views.build_day(cell, history, last, place, run)) else: - rf_token = f"{views.PAYLOAD_VER}:{hend}:{climate.recent_stamp(cid)}" + rf_token = views.recent_token(history, cid) slices["grade"] = slice_for( - "grade", f"{today.date().isoformat()}:14:14", rf_token, + "grade", views.grade_key(today, 14, 14), rf_token, lambda: views.build_grade(cell, today, 14, history, recent, cache_meta, place, run, after=14)) slices["forecast"] = slice_for( - "forecast", f"{today.date().isoformat()}:7", rf_token, + "forecast", views.forecast_key(today, 7), rf_token, lambda: views.build_forecast(cell, 7, history, recent, today, place, run)) - day_token = hist_token - if today > last: - day_token += f":h{int(time.time() // 3600)}" # obs from the hourly recent bundle slices["day"] = slice_for( - "day", today.date().isoformat(), day_token, + "day", views.day_key(today), views.day_token(history, today), lambda: views.build_day(cell, history, today, place, run, recent=recent)) # The bundle's identity is the combination of its slices' identities. diff --git a/migrate.py b/migrate.py index f879275..14f29e4 100644 --- a/migrate.py +++ b/migrate.py @@ -54,11 +54,11 @@ def migrate() -> int: skipped += 1 continue - token = f"{views.PAYLOAD_VER}:{views.hist_end(history)}" + token = views.history_token(history) start_ts, end_ts = views.cal_span(history, None, None, 24) - cal_key = f"{start_ts.date().isoformat()}:{end_ts.date().isoformat()}:24" + cal_key = views.calendar_key(start_ts, end_ts, 24) last = pd.Timestamp(history["date"].max()).normalize() - day_key = last.date().isoformat() + day_key = views.day_key(last) have_cal = store.get_payload("calendar", cell_id, cal_key, token) is not None have_day = store.get_payload("day", cell_id, day_key, token) is not None diff --git a/tests/test_api.py b/tests/test_api.py index bf35725..498e8f9 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -73,15 +73,45 @@ def test_day_detail_and_ladders(client, history): assert tmax["obs"]["grade"] -def test_day_maps_rate_limit_to_503(client, monkeypatch): +@pytest.mark.parametrize("path", ["grade", "calendar", "day", "forecast", "cell"]) +def test_every_data_route_maps_rate_limit_to_503(client, monkeypatch, path): def rate_limited(cell): raise RuntimeError("Open-Meteo request failed (429): rate-limited") monkeypatch.setattr(climate, "get_history", rate_limited) - r = client.get("/thermograph/api/v2/day", params=Q) + # A distinct spot per route: a store row cached by an earlier test would + # otherwise be checked only after the (failing) history fetch anyway. + r = client.get(f"/thermograph/api/v2/{path}", params={"lat": 51.5, "lon": -0.1}) assert r.status_code == 503 assert "rate-limited" in r.json()["detail"] +def test_cell_prefetch_never_fetches_upstream(client, monkeypatch): + def boom(cell): + raise AssertionError("prefetch=1 must never fetch weather upstream") + monkeypatch.setattr(climate, "get_history", boom) + monkeypatch.setattr(climate, "get_recent_forecast", boom) + monkeypatch.setattr(climate, "load_cached_history", lambda cell: None) + r = client.get("/thermograph/api/v2/cell", params={**Q, "prefetch": 1}) + assert r.status_code == 204 # cold cell: no body, no quota spent + + +def test_cell_prefetch_builds_history_slices_only(client): + r = client.get("/thermograph/api/v2/cell", params={"lat": 10.0, "lon": 10.0, "prefetch": 1}) + assert r.status_code == 200 + assert set(r.json()["slices"]) == {"calendar", "day"} + + +def test_calendar_placeless_payload_is_not_persisted(client, monkeypatch): + q = {"lat": -10.0, "lon": 20.0, "months": 2} + monkeypatch.setattr(climate, "reverse_geocode", lambda lat, lon: None) + r = client.get("/thermograph/api/v2/calendar", params=q) + assert r.status_code == 200 and r.json()["place"] is None + # The placeless payload wasn't cached, so the next request retries the + # label instead of replaying bare coordinates for the life of the token. + monkeypatch.setattr(climate, "reverse_geocode", lambda lat, lon: "Resolved, Now") + assert client.get("/thermograph/api/v2/calendar", params=q).json()["place"] == "Resolved, Now" + + def test_calendar_compact_range(client, history): r = client.get("/thermograph/api/v2/calendar", params={**Q, "months": 2}) assert r.status_code == 200 diff --git a/tests/test_views.py b/tests/test_views.py index c4d54e7..62d42ee 100644 --- a/tests/test_views.py +++ b/tests/test_views.py @@ -24,6 +24,32 @@ def test_views_and_migrate_import_without_the_web_stack(): subprocess.run([sys.executable, "-c", code], cwd=backend, check=True) +# ---- cache identity -------------------------------------------------------------- +# These formats are shared by the endpoints, the /cell bundle and migrate; pin +# them so a change is always deliberate (an accidental drift silently strands +# every existing derived row — change them only alongside a PAYLOAD_VER bump). + +def test_cache_identity_formats_are_pinned(history): + t = pd.Timestamp("2026-06-15") + assert views.grade_key(t, 14, 7) == "2026-06-15:14:7" + assert views.calendar_key(t, pd.Timestamp("2026-06-20"), 24) == "2026-06-15:2026-06-20:24" + assert views.day_key(t) == "2026-06-15" + assert views.forecast_key(t, 7) == "2026-06-15:7" + assert views.history_token(history) == f"{views.PAYLOAD_VER}:{views.hist_end(history)}" + + +def test_recent_token_composes_history_and_stamp(history, monkeypatch): + monkeypatch.setattr(climate, "recent_stamp", lambda cid: "stamp") + assert views.recent_token(history, "1_2") == f"{views.history_token(history)}:stamp" + + +def test_day_token_expires_hourly_only_beyond_the_archive(history): + last = pd.Timestamp(history["date"].max()).normalize() + assert views.day_token(history, last) == views.history_token(history) + future = views.day_token(history, last + pd.Timedelta(days=1)) + assert future.startswith(views.history_token(history) + ":h") + + # ---- cal_span clamping --------------------------------------------------------- def _hist(start, end): diff --git a/views.py b/views.py index 767f55a..c1f6b0d 100644 --- a/views.py +++ b/views.py @@ -8,6 +8,7 @@ semantics (audit runs, derived-store lookups, ETags) stay with the callers; ``run`` is an audit.RunAudit or None. """ import contextlib +import time import pandas as pd @@ -46,6 +47,50 @@ def hist_end(history) -> str: return pd.Timestamp(history["date"].max()).date().isoformat() +# --- cache identity ------------------------------------------------------------ +# The derived store caches payloads under (kind, cell, key) guarded by a validity +# token (see store.py). These functions are the ONLY definitions of each kind's +# key/token format: the per-view endpoints, the /cell bundle and the migrate +# script all derive cache identity here. (Formats drifting apart would silently +# split the cache — endpoints missing rows the bundle wrote, migrate +# materializing rows nobody reads.) + +def history_token(history) -> str: + """Validity for payloads derived from the archive record alone.""" + return f"{PAYLOAD_VER}:{hist_end(history)}" + + +def recent_token(history, cell_id: str) -> str: + """Validity for payloads that also grade the hourly recent/forecast bundle.""" + return f"{history_token(history)}:{climate.recent_stamp(cell_id)}" + + +def grade_key(target, days: int, after: int) -> str: + return f"{target.date().isoformat()}:{days}:{after}" + + +def calendar_key(start_ts, end_ts, months: int) -> str: + return f"{start_ts.date().isoformat()}:{end_ts.date().isoformat()}:{months}" + + +def day_key(target) -> str: + return target.date().isoformat() + + +def day_token(history, target) -> str: + """A fully-archived day stays valid until the record itself advances; a newer + day's observation comes from the hourly recent bundle, so its payload expires + on that bundle's own cadence (hourly).""" + token = history_token(history) + if target > pd.Timestamp(history["date"].max()).normalize(): + token += f":h{int(time.time() // 3600)}" + return token + + +def forecast_key(today, days: int) -> str: + return f"{today.date().isoformat()}:{days}" + + def _obs_from_row(row) -> dict: return {k: row[k] for k in OBS_COLS if k in row}