Single-source cache identity; shared fetch preamble and cache flow (#43)

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.
This commit is contained in:
Emi Griffith 2026-07-11 12:49:15 -07:00 committed by GitHub
parent 38a39df6ab
commit 7989f142a1
5 changed files with 198 additions and 137 deletions

224
app.py
View file

@ -5,7 +5,6 @@ import functools
import hashlib import hashlib
import json import json
import os import os
import time
import pandas as pd import pandas as pd
from fastapi import APIRouter, FastAPI, HTTPException, Query, Request, Response 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) 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 ---------------------------------------------------------------- # --- endpoints ----------------------------------------------------------------
def api_geocode(q: str = Query(..., min_length=1)): def api_geocode(q: str = Query(..., min_length=1)):
@ -255,32 +299,12 @@ def api_grade(
recent_days=days, recent_days=days,
cell_id=cell["id"], cell_id=cell["id"],
) as run: ) as run:
try: history, cache_meta, recent = _fetch_history(run, cell, recent_too=True)
with run.phase("history"): return _cached_response(
history, cache_meta = climate.get_history(cell) request, run, "grade", cell,
with run.phase("recent"): views.grade_key(target, days, after), views.recent_token(history, cell["id"]),
recent = climate.get_recent_forecast(cell) lambda place: views.build_grade(cell, target, days, history, recent,
except Exception as e: # noqa: BLE001 cache_meta, place, run, after=after))
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)
def api_calendar( def api_calendar(
@ -310,39 +334,22 @@ def api_calendar(
recent_days=months * 31, # approximate span graded recent_days=months * 31, # approximate span graded
cell_id=cell["id"], cell_id=cell["id"],
) as run: ) as run:
try: history, cache_meta, _ = _fetch_history(run, cell)
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.")
start_ts, end_ts = views.cal_span(history, start, end, months) 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"): def build(place):
place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) payload = views.build_calendar(cell, history, start_ts, end_ts, months, place, run)
payload = views.build_calendar(cell, history, start_ts, end_ts, months, place, run) full = not cache_meta.get("cached", False)
full = not cache_meta.get("cached", False) run.set(run_type="full" if full else "partial",
run.set(run_type="full" if full else "partial", history_source="fetch" if full else "cache")
history_source="fetch" if full else "cache") return payload
# Don't persist a payload whose place failed to resolve (a transient reverse-
# geocode miss) — otherwise the bare-coordinates fallback would stick for the # Compare loads several cells at once, so a transient reverse-geocode miss
# life of the token. Compare loads several cells at once, so this is where a # is most likely here; cache_placeless=False keeps that miss un-persisted.
# miss is most likely; leaving it uncached lets the next request retry. return _cached_response(request, run, "calendar", cell,
return _json_response( views.calendar_key(start_ts, end_ts, months),
store.put_payload("calendar", cell["id"], key, token, payload, views.history_token(history), build,
cache=place is not None), etag) cache_placeless=False)
def api_day( def api_day(
@ -367,38 +374,21 @@ def api_day(
lon=round(lon, 4), lon=round(lon, 4),
cell_id=cell["id"], cell_id=cell["id"],
) as run: ) as run:
try: history, cache_meta, _ = _fetch_history(run, cell)
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.")
last = pd.Timestamp(history["date"].max()).normalize() last = pd.Timestamp(history["date"].max()).normalize()
target = pd.Timestamp(date).normalize() if date else last target = pd.Timestamp(date).normalize() if date else last
run.set(target_date=target.date().isoformat()) run.set(target_date=target.date().isoformat())
key = target.date().isoformat() def build(place):
token = f"{views.PAYLOAD_VER}:{views.hist_end(history)}" payload = views.build_day(cell, history, target, place, run)
if target > last: full = not cache_meta.get("cached", False)
token += f":h{int(time.time() // 3600)}" # obs from the hourly recent bundle run.set(run_type="full" if full else "partial",
etag = _etag_for("day", cell["id"], key, token) history_source="fetch" if full else "cache")
if _not_modified(request, etag): return payload
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)
with run.phase("reverse_geocode"): return _cached_response(request, run, "day", cell,
place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) views.day_key(target), views.day_token(history, target),
payload = views.build_day(cell, history, target, place, run) build)
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)
def api_forecast( def api_forecast(
@ -419,34 +409,17 @@ def api_forecast(
with audit.RunAudit(endpoint="forecast", lat=round(lat, 4), lon=round(lon, 4), with audit.RunAudit(endpoint="forecast", lat=round(lat, 4), lon=round(lon, 4),
recent_days=days, cell_id=cell["id"]) as run: recent_days=days, cell_id=cell["id"]) as run:
try: history, _, fc = _fetch_history(run, cell, recent_too=True, recent_phase="forecast")
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.")
today = pd.Timestamp(datetime.date.today()) 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"): def build(place):
place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) payload = views.build_forecast(cell, days, history, fc, today, place, run)
payload = views.build_forecast(cell, days, history, fc, today, place, run) run.set(run_type="partial")
run.set(run_type="partial") return payload
return _json_response(store.put_payload("forecast", cell["id"], key, token, payload), etag)
return _cached_response(request, run, "forecast", cell,
views.forecast_key(today, days),
views.recent_token(history, cell["id"]), build)
def api_cell( def api_cell(
@ -485,18 +458,9 @@ def api_cell(
return Response(status_code=204) return Response(status_code=204)
cache_meta = {"cached": True} cache_meta = {"cached": True}
else: else:
try: history, cache_meta, recent = _fetch_history(run, cell, recent_too=True)
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.")
cid = cell["id"] cid = cell["id"]
hend = views.hist_end(history)
last = pd.Timestamp(history["date"].max()).normalize() last = pd.Timestamp(history["date"].max()).normalize()
with run.phase("reverse_geocode"): with run.phase("reverse_geocode"):
@ -513,34 +477,30 @@ def api_cell(
return {"etag": etag, "data": data} return {"etag": etag, "data": data}
slices = {} 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 # Calendar: the default last-24-months span — what the calendar view (and
# the cross-view prefetch) requests first. # the cross-view prefetch) requests first.
start_ts, end_ts = views.cal_span(history, None, None, 24) 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( 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)) lambda: views.build_calendar(cell, history, start_ts, end_ts, 24, place, run))
if prefetch: if prefetch:
# Latest archived day: its observation comes from history alone, so # Latest archived day: its observation comes from history alone, so
# it's buildable without the (skipped) recent bundle. # it's buildable without the (skipped) recent bundle.
slices["day"] = slice_for( 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)) lambda: views.build_day(cell, history, last, place, run))
else: else:
rf_token = f"{views.PAYLOAD_VER}:{hend}:{climate.recent_stamp(cid)}" rf_token = views.recent_token(history, cid)
slices["grade"] = slice_for( 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)) lambda: views.build_grade(cell, today, 14, history, recent, cache_meta, place, run, after=14))
slices["forecast"] = slice_for( 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)) 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( 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)) lambda: views.build_day(cell, history, today, place, run, recent=recent))
# The bundle's identity is the combination of its slices' identities. # The bundle's identity is the combination of its slices' identities.

View file

@ -54,11 +54,11 @@ def migrate() -> int:
skipped += 1 skipped += 1
continue 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) 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() 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_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 have_day = store.get_payload("day", cell_id, day_key, token) is not None

View file

@ -73,15 +73,45 @@ def test_day_detail_and_ladders(client, history):
assert tmax["obs"]["grade"] 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): def rate_limited(cell):
raise RuntimeError("Open-Meteo request failed (429): rate-limited") raise RuntimeError("Open-Meteo request failed (429): rate-limited")
monkeypatch.setattr(climate, "get_history", 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 r.status_code == 503
assert "rate-limited" in r.json()["detail"] 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): def test_calendar_compact_range(client, history):
r = client.get("/thermograph/api/v2/calendar", params={**Q, "months": 2}) r = client.get("/thermograph/api/v2/calendar", params={**Q, "months": 2})
assert r.status_code == 200 assert r.status_code == 200

View file

@ -24,6 +24,32 @@ def test_views_and_migrate_import_without_the_web_stack():
subprocess.run([sys.executable, "-c", code], cwd=backend, check=True) 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 --------------------------------------------------------- # ---- cal_span clamping ---------------------------------------------------------
def _hist(start, end): def _hist(start, end):

View file

@ -8,6 +8,7 @@ semantics (audit runs, derived-store lookups, ETags) stay with the callers;
``run`` is an audit.RunAudit or None. ``run`` is an audit.RunAudit or None.
""" """
import contextlib import contextlib
import time
import pandas as pd import pandas as pd
@ -46,6 +47,50 @@ def hist_end(history) -> str:
return pd.Timestamp(history["date"].max()).date().isoformat() 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: def _obs_from_row(row) -> dict:
return {k: row[k] for k in OBS_COLS if k in row} return {k: row[k] for k in OBS_COLS if k in row}