Jinemi/thermograph
4
0
Fork 0
Code Issues 2 Pull requests 3 Projects Releases Packages Wiki Activity Actions

Move the suggestion policy into places.py (#45)

Browse source 
api_suggest carried ~80 lines of ranking policy — the exactness-boosted
scoring, local/upstream blending with dedupe, and the token-respelling
correction loop — all consuming the match: prefix/fuzzy vocabulary that
places.py produces. Producer and consumer now live together:
places.suggest(q, upstream, limit) implements the whole policy with the
upstream geocoder lookup injected as a callable, so places stays free of
the fetch layer and the policy is testable without FastAPI.

The endpoint is the HTTP shim: call places.suggest, map the
nothing-at-all-to-serve failure to a 502. Behavior unchanged.

Policy tests: upstream skipped when the local answer convinces, blend +
dedupe, the exactness boost (a hamlet spelled like the typo can't beat
Seattle), single-typo queries answered by the fuzzy matcher without a
correction, the correction path verified via the upstream probe,
upstream failure degrading to local results, and the terminal
error-propagation case.
This commit is contained in:
Emi Griffith 2026-07-11 13:00:24 -07:00 committed by GitHub
parent 7c5b5137f1
commit b702e019d5
3 changed files with 157 additions and 68 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

76
app.py
View file

@ -179,76 +179,16 @@ def _suggest_upstream(q: str) -> tuple:
return tuple(climate.geocode(q, count=_SUGGEST_LIMIT))
def _suggest_score(r: dict, exact: bool) -> int:
"""Prominence rank with an exactness boost — big enough that a real place
beats a same-size typo match, small enough that a hamlet spelled exactly
like the typo can't outrank a metropolis one edit away ("Seatle", a
village, must not beat Seattle)."""
pop = r.get("population") or 0
return pop * 4 + 1 if exact else pop
def _suggest_merge(local: list, upstream: tuple, limit: int) -> list:
"""Blend local-index and upstream results into one top-`limit` list.
Everything the user's spelling matches exactly counts as exact; only the
local index's typo-tolerant matches take the fuzzy (unboosted) score."""
scored = [(r, _suggest_score(r, r.get("match") != "fuzzy")) for r in local]
scored += [(r, _suggest_score(r, True)) for r in upstream]
scored.sort(key=lambda t: t[1], reverse=True)
out, seen = [], set()
for r, _ in scored:
key = ((r.get("name") or "").casefold(), r.get("admin1") or "",
r.get("country_code") or "")
if key not in seen:
seen.add(key)
out.append(r)
if len(out) == limit:
break
return out
def api_suggest(q: str = Query(..., min_length=1)):
"""Type-ahead location suggestions: the top 5 places for a (possibly
typo'd) query prefix. The local GeoNames index answers instantly and
tolerates a single-letter typo; the upstream geocoder fills remaining slots
with what the index doesn't know (neighbourhoods, postcodes). When both
come up empty, one query token is respelled against known place-name tokens
and retried ("pest seattle" → "west seattle"); `corrected` reports the
respelling that produced the results."""
q = q.strip()
if len(q) < 2:
return {"results": [], "corrected": None}
local = places.search(q, _SUGGEST_LIMIT) # None while the index loads
results = list(local or [])
upstream_error = None
# Consult the upstream geocoder unless the local index already answered
# convincingly: full slots and a substantial place matching the spelling as
# typed. Fuzzy hits don't count as convincing — they're guesses, and when
# the query is an alternate name the index doesn't know ("münchen" prefix-
# matches only small towns; Munich lives upstream), neither those towns nor
# a coincidental fuzzy big-city hit may suppress the real answer.
prominent = max((r.get("population") or 0 for r in results
if r.get("match") == "prefix"), default=0)
if len(results) < _SUGGEST_LIMIT or prominent < 100_000:
try:
results = _suggest_merge(results, _suggest_upstream(q), _SUGGEST_LIMIT)
except Exception as e: # noqa: BLE001 - local results (if any) still serve
upstream_error = e
corrected = None
if not results and len(q) >= 4:
for phrase in places.corrections(q):
hits = places.search(phrase, _SUGGEST_LIMIT) or []
if not hits:
try:
hits = list(_suggest_upstream(phrase))
except Exception: # noqa: BLE001 - a failed probe just means no hits
hits = []
if hits:
results, corrected = hits[:_SUGGEST_LIMIT], phrase
break
if not results and local is None and upstream_error is not None:
raise HTTPException(status_code=502, detail=f"geocoding failed: {upstream_error}")
return {"results": results[:_SUGGEST_LIMIT], "corrected": corrected}
typo'd) query prefix; `corrected` reports the respelling that produced the
results ("pest seattle" → "west seattle"), or null. The ranking/typo policy
lives in places.suggest."""
try:
results, corrected = places.suggest(q, _suggest_upstream, _SUGGEST_LIMIT)
except Exception as e: # noqa: BLE001 - nothing at all to serve
raise HTTPException(status_code=502, detail=f"geocoding failed: {e}")
return {"results": results, "corrected": corrected}
def api_place(

77
places.py
View file

@ -234,6 +234,83 @@ def search(q: str, limit: int = 5) -> list[dict] | None:
return out
def _suggest_score(r: dict, exact: bool) -> int:
"""Prominence rank with an exactness boost — big enough that a real place
beats a same-size typo match, small enough that a hamlet spelled exactly
like the typo can't outrank a metropolis one edit away ("Seatle", a
village, must not beat Seattle)."""
pop = r.get("population") or 0
return pop * 4 + 1 if exact else pop
def _suggest_merge(local: list, upstream: tuple, limit: int) -> list:
"""Blend local-index and upstream results into one top-`limit` list.
Everything the user's spelling matches exactly counts as exact; only the
local index's typo-tolerant matches take the fuzzy (unboosted) score."""
scored = [(r, _suggest_score(r, r.get("match") != "fuzzy")) for r in local]
scored += [(r, _suggest_score(r, True)) for r in upstream]
scored.sort(key=lambda t: t[1], reverse=True)
out, seen = [], set()
for r, _ in scored:
key = ((r.get("name") or "").casefold(), r.get("admin1") or "",
r.get("country_code") or "")
if key not in seen:
seen.add(key)
out.append(r)
if len(out) == limit:
break
return out
def suggest(q: str, upstream, limit: int = 5) -> tuple[list[dict], str | None]:
"""The whole suggestion policy: the top `limit` places for a (possibly
typo'd) query prefix, plus the respelling that produced them (or None).
The local index answers instantly and tolerates a single-letter typo;
``upstream(q) -> iterable`` (an injected geocoder lookup, so this module
stays free of the fetch layer) fills remaining slots with what the index
doesn't know (neighbourhoods, postcodes). The upstream is consulted unless
the local index already answered convincingly: full slots and a substantial
place matching the spelling as typed. Fuzzy hits don't count as convincing —
they're guesses, and when the query is an alternate name the index doesn't
know ("münchen" prefix-matches only small towns; Munich lives upstream),
neither those towns nor a coincidental fuzzy big-city hit may suppress the
real answer. When both come up empty, one query token is respelled against
known place-name tokens and retried ("pest seattle" → "west seattle").
An upstream failure degrades to whatever the index found — it propagates
only when there is nothing at all to serve (index still loading, no
correction hits)."""
q = q.strip()
if len(q) < 2:
return [], None
local = search(q, limit) # None while the index loads
results = list(local or [])
upstream_error = None
prominent = max((r.get("population") or 0 for r in results
if r.get("match") == "prefix"), default=0)
if len(results) < limit or prominent < 100_000:
try:
results = _suggest_merge(results, tuple(upstream(q)), limit)
except Exception as e: # noqa: BLE001 - local results (if any) still serve
upstream_error = e
corrected = None
if not results and len(q) >= 4:
for phrase in corrections(q):
hits = search(phrase, limit) or []
if not hits:
try:
hits = list(upstream(phrase))
except Exception: # noqa: BLE001 - a failed probe just means no hits
hits = []
if hits:
results, corrected = hits[:limit], phrase
break
if not results and local is None and upstream_error is not None:
raise upstream_error
return results[:limit], corrected
def corrections(q: str, max_phrases: int = 3) -> list[str]:
"""Candidate respellings of `q` with one token replaced by a known
place-name token a single edit away — "pest seattle" → "west seattle".

72
tests/test_places.py
View file

@ -97,6 +97,78 @@ def test_search_result_shape(index):
"population": 2_100_000, "match": "prefix"}
# ---- the full suggestion policy ----------------------------------------------------
def _no_upstream(q):
raise AssertionError("upstream must not be consulted")
def test_suggest_skips_upstream_when_local_answer_convinces(index):
# Full slots (limit=1) + a prominent exact-prefix match: no upstream call.
results, corrected = places.suggest("paris", _no_upstream, limit=1)
assert [r["name"] for r in results] == ["Paris"] and corrected is None
def test_suggest_blends_upstream_and_dedupes(index):
munich = {"name": "Munich", "admin1": "Bavaria", "country": "Germany",
"country_code": "DE", "lat": 48.14, "lon": 11.58, "population": 1_500_000}
seattle_dup = {"name": "Seattle", "admin1": "Washington", "country": "United States",
"country_code": "US", "lat": 47.6, "lon": -122.33, "population": 737_015}
results, _ = places.suggest("sea", lambda q: [munich, seattle_dup], limit=5)
names = [r["name"] for r in results]
assert names.count("Seattle") == 1 # local + upstream copies merged
assert "Munich" in names # upstream filled a free slot
def test_suggest_exactness_boost_beats_fuzzy_size(index):
# "seatle" fuzzy-matches Seattle locally; an exact upstream hamlet named
# "Seatle" must not outrank it (pop*4+1 boost vs 750k unboosted).
hamlet = {"name": "Seatle", "admin1": None, "country": "Nowhere",
"country_code": "XX", "lat": 0.0, "lon": 0.0, "population": 500}
results, _ = places.suggest("seatle", lambda q: [hamlet], limit=5)
assert results[0]["name"] == "Seattle"
def test_suggest_single_typo_needs_no_correction(index):
# "pest seattle" is one edit from the "west seattle" prefix, so the fuzzy
# matcher answers directly — no respelling reported.
results, corrected = places.suggest("pest seattle", lambda q: [], limit=5)
assert corrected is None
assert results and results[0]["name"] == "West Seattle"
def test_suggest_corrections_kick_in_when_all_else_fails(index):
# "pest portland" is >1 edit from every indexed name, so search and the
# upstream probe both miss; the token respelling ("pest"→"west") is then
# verified against the upstream geocoder.
wp = {"name": "West Portland", "admin1": "Oregon", "country": "United States",
"country_code": "US", "lat": 45.48, "lon": -122.72, "population": 10_000}
def upstream(q):
return [wp] if q == "west portland" else []
results, corrected = places.suggest("pest portland", upstream, limit=5)
assert corrected == "west portland"
assert results and results[0]["name"] == "West Portland"
def test_suggest_upstream_failure_degrades_to_local(index):
def broken(q):
raise RuntimeError("upstream down")
results, _ = places.suggest("sea", broken, limit=5)
assert [r["name"] for r in results] == ["Seattle", "SeaTac"]
def test_suggest_raises_only_with_nothing_to_serve(monkeypatch):
monkeypatch.setattr(places, "_data", None) # index still loading
def broken(q):
raise RuntimeError("upstream down")
with pytest.raises(RuntimeError, match="upstream down"):
places.suggest("seattle", broken, limit=5)
def test_suggest_short_queries_answer_empty(index):
assert places.suggest("s", _no_upstream, limit=5) == ([], None)
# ---- corrections ------------------------------------------------------------------
def test_corrections_respell_a_typod_token(index):