From 4cbb008f82adc2bee50da9888cea4a85463f89cf Mon Sep 17 00:00:00 2001 From: Zachary Schwartz <112527479+ZachSchwartz@users.noreply.github.com> Date: Tue, 1 Sep 2026 15:22:37 -0500 Subject: [PATCH 1/7] initial commit --- .gitignore | 4 + src/api.py | 97 ++++++++++++++++- src/cache.py | 240 ++++++++++++++++++++++++++++++++++++++++++ src/census.py | 76 +++++++++---- src/geocoder.py | 47 +++++++-- src/google.py | 59 +++++++---- test/api_test.py | 28 +++-- test/cache_test.py | 201 +++++++++++++++++++++++++++++++++++ test/census_test.py | 48 +++++++++ test/geocoder_test.py | 98 +++++++++++++---- test/google_test.py | 46 ++++++++ 11 files changed, 858 insertions(+), 86 deletions(-) create mode 100644 src/cache.py create mode 100644 test/cache_test.py diff --git a/.gitignore b/.gitignore index cb1a262..6198ea5 100644 --- a/.gitignore +++ b/.gitignore @@ -218,3 +218,7 @@ __marimo__/ .streamlit/secrets.toml docs/ + +# Geocoder response caches: +*.sqlite +*.sqlite-journal diff --git a/src/api.py b/src/api.py index 2b9ad4c..3712fc9 100644 --- a/src/api.py +++ b/src/api.py @@ -13,10 +13,12 @@ from abc import ABC, abstractmethod from dataclasses import dataclass, field from enum import IntEnum -from typing import Any, Callable, Dict, List, Optional, Type +from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple, Type import requests +from .cache import Cache, normalize_query + KEY_ENV_VARS = { "geocodio": "GEOCODIO_API_KEY", "google": "GOOGLE_GEOCODING_API_KEY", @@ -140,7 +142,14 @@ def decorator(cls): class Provider(ABC): - """Base class for geocoding providers; subclasses self-register via @register.""" + """ + Base class for geocoding providers; subclasses self-register via @register. + + Subclasses supply three pieces and inherit the rest: ``_fetch`` calls the API, + ``parse`` turns one raw response into a result, and ``cache_key`` names the + query a record resolves to. Deduplication and caching are handled once here so + no provider has to repeat them. + """ name: str = "" requires_key: bool = False @@ -148,12 +157,92 @@ class Provider(ABC): RETRY_BACKOFF = 5 MAX_ACCURACY = AccuracyLevel.ROOFTOP - def __init__(self, api_key: Optional[str] = None): + def __init__(self, api_key: Optional[str] = None, cache: Optional[Cache] = None): self.api_key = api_key + self.cache = cache if cache is not None else Cache() + + def cache_key(self, record: SourceRecord) -> str: + """ + Returns the query text this provider would send for the given record. + + Two records with the same key are answered by a single API call, so the + key must cover everything that changes the response — the address + components a provider actually submits plus any pinned request parameters + that steer the result. Providers whose query is not the full address + override this. + + Parameters + ---------- + record : SourceRecord + The record whose query is being composed. + + Return + ---------- + str + The query text identifying this record's response. + """ + return record.address_string() + + @abstractmethod + def _fetch(self, records: List[SourceRecord]) -> Iterator[Tuple[SourceRecord, Dict[str, Any]]]: + """ + Yields each record paired with its raw provider response, as replies arrive. + + Yielding rather than returning lets the caller cache responses at whatever + granularity the provider produces them, so an interrupted run keeps the + calls it already paid for. Only responses that have been validated against + the provider's expected shape may be yielded, because everything yielded is + cached; a malformed or failed response must raise instead. + """ @abstractmethod + def parse(self, raw: Dict[str, Any]) -> GeocodeResult: + """ + Converts one raw provider response into a normalized, graded GeocodeResult. + + Cached and freshly fetched responses both come through here, so grading + stays live: a stored response is always scored by the current rules rather + than the ones in force when it was fetched. This must be a pure mapping + over the response and must not call the API. + """ + def geocode(self, records: List[SourceRecord]) -> List[GeocodeResult]: - """Returns one GeocodeResult per input record, in the same order.""" + """ + Returns one GeocodeResult per input record, in the same order. + + Records sharing a query are collapsed to a single API call, and any query + already in the cache skips the API entirely, so a list of 5000 addresses + costs one call per distinct address never seen before. Deduplication + applies even with no cache file configured. + + Parameters + ---------- + records : List[SourceRecord] + The source rows to geocode. + + Return + ---------- + List[GeocodeResult] + One result per input record, aligned by position. + """ + queries = [self.cache_key(record) for record in records] + keys = [normalize_query(query) for query in queries] + responses = self.cache.lookup(self.name, keys) + + pending: Dict[str, SourceRecord] = {} + for key, record in zip(keys, records): + if key not in responses and key not in pending: + pending[key] = record + + if pending: + for record, raw in self._fetch(list(pending.values())): + query = self.cache_key(record) + key = normalize_query(query) + responses[key] = raw + self.cache.store(self.name, key, query, raw) + self.cache.commit() + + return [self.parse(responses[key]) if key in responses else GeocodeResult(match_notes="No match") for key in keys] def _request_with_retry(self, send: Callable[[], requests.Response]) -> requests.Response: """ diff --git a/src/cache.py b/src/cache.py new file mode 100644 index 0000000..62b8cf9 --- /dev/null +++ b/src/cache.py @@ -0,0 +1,240 @@ +#!/usr/bin/python3 +# -.- coding: utf-8 -.- +# -.- dependencies: Python 3.8+ -.- + +""" +Geocoder — provider response cache + +Copyright (c) 2026 Pangaea Information Technologies, Ltd. +""" + +import json +import sqlite3 +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, Iterable, Iterator, List, Optional, Sequence + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS cache ( + api TEXT NOT NULL, + query TEXT NOT NULL, + original_query TEXT NOT NULL, + response TEXT NOT NULL, + fetched_at TEXT NOT NULL, + PRIMARY KEY (api, query) +) +""" + + +def normalize_query(query: str) -> str: + """ + Canonicalizes a provider query into the key its response is stored under. + + Geocoders ignore letter case and runs of whitespace, so folding both lets + "123 Main St" and "123 MAIN ST" share a single cached response rather than + costing two identical API calls. Nothing else is altered: punctuation and + abbreviations can change what a geocoder returns, so they are left alone. + + Parameters + ---------- + query : str + The query text a provider would send for a record. + + Return + ---------- + str + The canonical key for that query. + """ + return " ".join(query.split()).casefold() + + +class Cache: + """ + A SQLite-backed store of raw provider responses keyed by the query that produced them. + + The cache is never a source of truth. It holds only what a provider actually + returned, so deleting any or all of its files changes nothing about the output + beyond how many API calls a run costs. Results are parsed from the stored + response on every run, which keeps a cached row scored by the current grading + rules rather than the ones in force when it was fetched. + + One table serves every provider, so adding a provider needs no schema change + and consolidating two files is a single statement:: + + ATTACH 'other.sqlite' AS other; + INSERT OR IGNORE INTO cache SELECT * FROM other.cache; + + A cache with no writable path and no readers is a working no-op, which lets + providers use one unconditionally. + """ + + COMMIT_INTERVAL = 250 + LOOKUP_CHUNK = 500 + + def __init__(self, path: Optional[str] = None, read_paths: Sequence[str] = ()): + self._writer = self._open_writer(path) if path else None + self._readers = [self._open_reader(read_path) for read_path in read_paths] + self._uncommitted = 0 + + @staticmethod + def _open_writer(path: str) -> sqlite3.Connection: + """ + Opens the read-write cache, creating the file and table when absent. + + Parameters + ---------- + path : str + The path of the cache file to open or create. + + Return + ---------- + sqlite3.Connection + A connection with the cache table in place. + + Raises + ---------- + ValueError + If the path names an existing file that is not a SQLite database. + """ + connection = sqlite3.connect(path) + try: + connection.execute(SCHEMA) + connection.commit() + except sqlite3.DatabaseError as error: + connection.close() + raise ValueError(f"'{path}' cannot be used as a cache: {error}") from error + return connection + + @staticmethod + def _open_reader(path: str) -> sqlite3.Connection: + """ + Opens a cache file read-only and confirms it carries the cache table. + + Parameters + ---------- + path : str + The path of an existing cache file. + + Return + ---------- + sqlite3.Connection + A connection that cannot modify the file. + + Raises + ---------- + ValueError + If the file is not a SQLite database or holds no cache table. + """ + connection = sqlite3.connect(f"{Path(path).resolve().as_uri()}?mode=ro", uri=True) + try: + found = connection.execute("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'cache'").fetchone() + except sqlite3.DatabaseError as error: + connection.close() + raise ValueError(f"'{path}' is not a readable SQLite database: {error}") from error + + if found is None: + connection.close() + raise ValueError(f"'{path}' is not a geocoder cache (it has no 'cache' table)") + return connection + + def _connections(self) -> Iterator[sqlite3.Connection]: + """Yields the writable cache first, then each read-only cache in the order given.""" + if self._writer is not None: + yield self._writer + yield from self._readers + + def lookup(self, api: str, keys: Iterable[str]) -> Dict[str, Any]: + """ + Fetches the stored responses for the given keys, writable cache first. + + Each key is looked up in the writable cache and then in each read-only + cache, and the first file holding it wins. Keys with no stored response + are absent from the result. Lookups are chunked so a run of any size stays + within the SQLite limit on bound parameters. + + Parameters + ---------- + api : str + The provider name whose entries are searched. + keys : Iterable[str] + The normalized query keys to look for. + + Return + ---------- + Dict[str, Any] + Each found key mapped to its decoded response. + """ + found: Dict[str, Any] = {} + outstanding = list(dict.fromkeys(keys)) + + for connection in self._connections(): + if not outstanding: + break + for start in range(0, len(outstanding), self.LOOKUP_CHUNK): + chunk = outstanding[start : start + self.LOOKUP_CHUNK] + placeholders = ",".join("?" * len(chunk)) + rows = connection.execute(f"SELECT query, response FROM cache WHERE api = ? AND query IN ({placeholders})", [api, *chunk]) + for key, response in rows: + found[key] = json.loads(response) + outstanding = [key for key in outstanding if key not in found] + + return found + + def store(self, api: str, key: str, query: str, response: Any) -> None: + """ + Records one provider response, committing once a batch has accumulated. + + Committing as the run proceeds means a crash costs only the calls made + since the last commit rather than every call made so far. Nothing is + written when no writable cache is configured. + + Parameters + ---------- + api : str + The provider name the response came from. + key : str + The normalized query key to store the response under. + query : str + The query text as the provider composed it, kept for manual review. + response : Any + The raw provider response, stored as JSON. + """ + if self._writer is None: + return + + fetched_at = datetime.now(timezone.utc).isoformat(timespec="seconds") + self._writer.execute( + "INSERT OR REPLACE INTO cache (api, query, original_query, response, fetched_at) VALUES (?, ?, ?, ?, ?)", + (api, key, query, json.dumps(response, ensure_ascii=False), fetched_at), + ) + + self._uncommitted += 1 + if self._uncommitted >= self.COMMIT_INTERVAL: + self.commit() + + def commit(self) -> None: + """Flushes any stored responses that have not yet been committed.""" + if self._writer is not None and self._uncommitted: + self._writer.commit() + self._uncommitted = 0 + + def close(self) -> None: + """Commits outstanding writes and closes every open cache file.""" + self.commit() + for connection in self._connections(): + connection.close() + self._writer = None + self._readers = [] + + def __enter__(self) -> "Cache": + """Returns the cache so it can be used as a context manager.""" + return self + + def __exit__(self, *_exception) -> None: + """Closes the cache when the context exits.""" + self.close() + + +def missing_files(paths: Sequence[str]) -> List[str]: + """Returns the subset of the given paths that are not existing files.""" + return [path for path in paths if not Path(path).is_file()] diff --git a/src/census.py b/src/census.py index 49da47d..bada20e 100644 --- a/src/census.py +++ b/src/census.py @@ -10,7 +10,7 @@ import csv import io -from typing import Dict, List +from typing import Any, Dict, Iterator, List, Tuple import requests @@ -54,9 +54,25 @@ class CensusProvider(Provider): "block", ] - def geocode(self, records: List[SourceRecord]) -> List[GeocodeResult]: + def cache_key(self, record: SourceRecord) -> str: """ - Geocodes records in batches and returns one result per record, in order. + Builds the key from the components Census is sent, plus the pinned dataset. + + Country is excluded because the addressbatch CSV has no country column, so + two rows differing only in country resolve to the same request. The + benchmark and vintage are included because they select which dataset + answers the query, and moving off Census2020 would make stored responses + answers to a different question. + """ + parts = [record.address, record.city, record.stateprov, record.postalcode, self.BENCHMARK, self.VINTAGE] + return "|".join(parts) + + def _fetch(self, records: List[SourceRecord]) -> Iterator[Tuple[SourceRecord, Dict[str, Any]]]: + """ + Posts records in batches, yielding each record with its response row. + + Yielding per batch rather than per run means an interrupted job keeps + every batch that already came back. Parameters ---------- @@ -65,23 +81,24 @@ def geocode(self, records: List[SourceRecord]) -> List[GeocodeResult]: Return ---------- - List[GeocodeResult] - One result per input record, aligned by position. + Iterator[Tuple[SourceRecord, Dict[str, Any]]] + Each record paired with its raw Census response row. """ - results_by_key: Dict[int, GeocodeResult] = {} for start in range(0, len(records), self.BATCH_SIZE): - self._geocode_batch(records[start : start + self.BATCH_SIZE], results_by_key) + yield from self._fetch_batch(records[start : start + self.BATCH_SIZE]) - return [results_by_key.get(record.internal_key, GeocodeResult(match_notes="No match")) for record in records] - - def _geocode_batch(self, batch: List[SourceRecord], results_by_key: Dict[int, GeocodeResult]) -> None: - """Posts one CSV batch, verifies its row count, and stores each result by internal key.""" + def _fetch_batch(self, batch: List[SourceRecord]) -> Iterator[Tuple[SourceRecord, Dict[str, Any]]]: + """Posts one CSV batch, verifies its row count, and matches rows back by internal key.""" response = self._post_batch(batch) rows = [row for row in csv.reader(io.StringIO(response.text)) if row] if len(rows) != len(batch): raise ValueError(f"Census returned {len(rows)} rows for {len(batch)} submitted records; the service or benchmark may have changed") + + records_by_key = {record.internal_key: record for record in batch} for row in rows: - results_by_key[int(row[0])] = self._parse_row(row) + record = records_by_key.get(int(row[0])) + if record is not None: + yield record, self._build_raw(row) def _post_batch(self, batch: List[SourceRecord]) -> requests.Response: """Posts one CSV batch through the retrying request helper.""" @@ -112,27 +129,44 @@ def _build_csv(batch: List[SourceRecord]) -> str: ) return buffer.getvalue() - def _parse_row(self, row: List[str]) -> GeocodeResult: + def _build_raw(self, row: List[str]) -> Dict[str, Any]: + """ + Names the positional fields of one response row, validating a match first. + + Reading the headerless response by position is only safe while the pinned + layout holds, so a match row is checked here, where it arrives from the + network, and never again. Everything downstream reads the named fields. + + The echoed id is dropped: it is this run's internal key, which carries no + meaning once the response outlives the run that fetched it. + """ + status = row[2] if len(row) > 2 else "No_Match" + if status == "Match": + self._check_layout(row) + + raw = dict(zip(self.RESPONSE_FIELDS, row)) + raw.pop("id", None) + return raw + + def parse(self, raw: Dict[str, Any]) -> GeocodeResult: """ Converts one Census response row into a normalized GeocodeResult. Accuracy is graded from the populated result fields rather than the Census match type, so it reflects how specific the returned location is. """ - result = self._build_result(row) + result = self._build_result(raw) result.accuracy = grade_accuracy(result, self.MAX_ACCURACY) return result - def _build_result(self, row: List[str]) -> GeocodeResult: + def _build_result(self, raw: Dict[str, Any]) -> GeocodeResult: """Maps a Census response row to a result without scoring its accuracy.""" - raw = dict(zip(self.RESPONSE_FIELDS, row)) - status = row[2] if len(row) > 2 else "No_Match" + status = raw.get("match_status", "No_Match") if status == "Match": - self._check_layout(row) - exact = row[3].strip().lower() == "exact" - address, city, stateprov, postalcode = self._split_address(row[4]) - longitude, latitude = self._split_coordinates(row[5]) + exact = raw.get("match_type", "").strip().lower() == "exact" + address, city, stateprov, postalcode = self._split_address(raw.get("matched_address", "")) + longitude, latitude = self._split_coordinates(raw.get("coordinates", "")) return GeocodeResult( result_id=raw.get("tigerline_id", ""), result_address=address, diff --git a/src/geocoder.py b/src/geocoder.py index 0729de3..395fbfb 100644 --- a/src/geocoder.py +++ b/src/geocoder.py @@ -23,6 +23,7 @@ SourceRecord, resolve_api_key, ) +from .cache import Cache, missing_files CANONICAL_FIELDS = [ "ID", @@ -332,6 +333,20 @@ def parse_args(argv: Optional[List[str]] = None): action="store_true", help="use the worksheet name as COUNTRY when a row's country is blank", ) + parser.add_argument( + "--cache", + default=None, + metavar="FILE", + help="SQLite cache of prior API responses, read and written (created if absent)", + ) + parser.add_argument( + "--cacheRead", + dest="cache_read", + action="append", + default=[], + metavar="FILE", + help="existing SQLite cache to read but never write; repeatable", + ) parser.add_argument( "--debug", action="store_true", @@ -352,8 +367,9 @@ def main(argv: Optional[List[str]] = None) -> None: Raises ---------- SystemExit - If the input file is missing, the output file already exists, the api is - unknown, or a required API key is not configured. + If the input file is missing, the output file already exists, a read-only + cache is missing or unusable, the api is unknown, or a required API key is + not configured. """ load_dotenv() args = parse_args(argv) @@ -363,6 +379,10 @@ def main(argv: Optional[List[str]] = None) -> None: if os.path.exists(args.outfile): raise SystemExit(f"error: output file '{args.outfile}' already exists") + absent = missing_files(args.cache_read) + if absent: + raise SystemExit(f"error: read-only cache file(s) do not exist: {', '.join(absent)}") + provider_cls = PROVIDERS.get(args.api) if provider_cls is None: available = ", ".join(sorted(PROVIDERS)) or "none" @@ -374,15 +394,20 @@ def main(argv: Optional[List[str]] = None) -> None: if not api_key: raise SystemExit(f"error: api '{args.api}' requires an API key " f"(pass --apiKey or set {KEY_ENV_VARS[args.api]})") - provider = provider_cls(api_key) - process_workbook( - args.infile, - args.outfile, - provider, - worksheet=args.worksheet, - country_per_sheet=args.country_per_sheet, - debug=args.debug, - ) + try: + cache = Cache(args.cache, args.cache_read) + except ValueError as error: + raise SystemExit(f"error: {error}") from error + + with cache: + process_workbook( + args.infile, + args.outfile, + provider_cls(api_key, cache), + worksheet=args.worksheet, + country_per_sheet=args.country_per_sheet, + debug=args.debug, + ) print(f"wrote {args.outfile}") diff --git a/src/google.py b/src/google.py index 0547f5e..ea77e58 100644 --- a/src/google.py +++ b/src/google.py @@ -8,7 +8,7 @@ Copyright (c) 2026 Pangaea Information Technologies, Ltd. """ -from typing import Dict, List +from typing import Any, Dict, Iterator, List, Tuple import requests @@ -54,9 +54,13 @@ class GoogleProvider(Provider): "country": "result_country", } - def geocode(self, records: List[SourceRecord]) -> List[GeocodeResult]: + def _fetch(self, records: List[SourceRecord]) -> Iterator[Tuple[SourceRecord, Dict[str, Any]]]: """ - Geocodes each record with a single request and returns results in order. + Queries one address per request, yielding each response as it arrives. + + A status other than a match or an empty result means the request itself + failed — a rejected key or an exhausted quota — so it is raised here rather + than yielded, keeping a failure out of the cache and off the output. Parameters ---------- @@ -65,21 +69,20 @@ def geocode(self, records: List[SourceRecord]) -> List[GeocodeResult]: Return ---------- - List[GeocodeResult] - One result per input record, aligned by position. - """ - return [self._geocode_one(record) for record in records] - - def _geocode_one(self, record: SourceRecord) -> GeocodeResult: - """Queries one address and grades the parsed response by its location_type.""" - payload = self._request(record.address_string()) - status = payload.get("status", "UNKNOWN") + Iterator[Tuple[SourceRecord, Dict[str, Any]]] + Each record paired with its raw Google response. - if status == "OK": - return self._parse_result(payload["results"][0]) - if status == "ZERO_RESULTS": - return GeocodeResult(match_notes="No match", raw=payload) - raise ValueError(f"Google geocoding failed with status {status!r}: {payload.get('error_message', '')}".strip()) + Raises + ---------- + ValueError + If Google reports a status other than OK or ZERO_RESULTS. + """ + for record in records: + payload = self._request(record.address_string()) + status = payload.get("status", "UNKNOWN") + if status not in ("OK", "ZERO_RESULTS"): + raise ValueError(f"Google geocoding failed with status {status!r}: {payload.get('error_message', '')}".strip()) + yield record, payload def _request(self, address: str) -> Dict: """Sends one geocoding request through the retrying request helper.""" @@ -92,21 +95,31 @@ def _request(self, address: str) -> Dict: ) return response.json() - def _parse_result(self, match: Dict) -> GeocodeResult: + def parse(self, raw: Dict[str, Any]) -> GeocodeResult: """ - Converts one Google result into a normalized GeocodeResult. + Converts one Google response into a normalized GeocodeResult. The graded accuracy is capped at the tier implied by ``location_type`` so an interpolated or centroid match cannot report rooftop precision on the strength of the echoed address fields. """ - result = self._build_result(match) + if raw.get("status") != "OK": + return GeocodeResult(match_notes="No match", raw=raw) + + result = self._build_result(raw) cap = self.LOCATION_TYPE_ACCURACY.get(result.location_type, AccuracyLevel.NONE) result.accuracy = grade_accuracy(result, cap) return result - def _build_result(self, match: Dict) -> GeocodeResult: - """Maps a Google result to a GeocodeResult without scoring its accuracy.""" + def _build_result(self, raw: Dict[str, Any]) -> GeocodeResult: + """ + Maps the best Google match to a GeocodeResult without scoring its accuracy. + + The whole response envelope is kept as the raw value rather than the single + match it was read from, so what is cached and reported is what Google + actually said. + """ + match = raw["results"][0] components = self._extract_components(match) geometry = match.get("geometry", {}) location = geometry.get("location", {}) @@ -122,7 +135,7 @@ def _build_result(self, match: Dict) -> GeocodeResult: longitude=str(location.get("lng", "")), match_type="partial" if match.get("partial_match") else "exact", location_type=geometry.get("location_type", ""), - raw=match, + raw=raw, ) def _extract_components(self, match: Dict) -> Dict[str, str]: diff --git a/test/api_test.py b/test/api_test.py index 274712e..df9cfdf 100644 --- a/test/api_test.py +++ b/test/api_test.py @@ -19,9 +19,13 @@ def test_register_adds_to_registry(): @register("temp_provider") class _Temp(Provider): - def geocode(self, records): - """Returns no results; the class only exercises registration.""" - return [] + def _fetch(self, records): + """Yields nothing; the class only exercises registration.""" + return iter(()) + + def parse(self, raw): + """Returns an empty result; the class only exercises registration.""" + return GeocodeResult(raw=raw) try: assert api.PROVIDERS["temp_provider"] is _Temp @@ -78,14 +82,22 @@ def raise_for_status(self): class _RetryProvider(Provider): - """Routes a single retried request through geocode for the retry tests.""" + """Routes a single retried request through the provider for the retry tests.""" def __init__(self, send): super().__init__() self.send = send - def geocode(self, records): - """Returns the response from one retried request, ignoring records.""" + def _fetch(self, records): + """Yields nothing; the class only exercises the retry helper.""" + return iter(()) + + def parse(self, raw): + """Returns an empty result; the class only exercises the retry helper.""" + return GeocodeResult(raw=raw) + + def send_once(self): + """Returns the response from one retried request.""" return self._request_with_retry(self.send) @@ -100,7 +112,7 @@ def send(): raise api.requests.ConnectionError("connection reset") return _FakeResponse("ok") - response = _RetryProvider(send).geocode([]) + response = _RetryProvider(send).send_once() assert attempts["count"] == 3 assert response.text == "ok" @@ -116,5 +128,5 @@ def send(): raise api.requests.ConnectionError("connection reset") with pytest.raises(api.requests.ConnectionError): - _RetryProvider(send).geocode([]) + _RetryProvider(send).send_once() assert attempts["count"] == Provider.MAX_ATTEMPTS diff --git a/test/cache_test.py b/test/cache_test.py new file mode 100644 index 0000000..d4caa14 --- /dev/null +++ b/test/cache_test.py @@ -0,0 +1,201 @@ +#!/usr/bin/python3 +# -.- coding: utf-8 -.- +# -.- dependencies: Python 3.8+ -.- + +""" +Geocoder Cache Tests + +Copyright (c) 2026 Pangaea Information Technologies, Ltd. +""" + +import sqlite3 + +import pytest + +from src.api import GeocodeResult, Provider, SourceRecord +from src.cache import Cache, missing_files, normalize_query + + +class _CountingProvider(Provider): + """Counts the records it is asked to fetch so cache hits can be observed.""" + + name = "counting" + + def __init__(self, cache=None): + super().__init__(cache=cache) + self.fetched = [] + + def _fetch(self, records): + """Yields a response per record and records which records were requested.""" + for record in records: + self.fetched.append(record.address_string()) + yield record, {"echo": record.address_string()} + + def parse(self, raw): + """Returns a result carrying the echoed query so hits can be identified.""" + return GeocodeResult(result_address=raw["echo"], raw=raw) + + +def _record(key, address, city="Town", stateprov="CA"): + """Builds a SourceRecord with the given internal key and address.""" + return SourceRecord(internal_key=key, address=address, city=city, stateprov=stateprov) + + +def test_normalize_query_folds_case_and_whitespace(): + """Case and runs of whitespace collapse so equivalent queries share one key.""" + assert normalize_query("123 Main St") == normalize_query("123 MAIN ST") + assert normalize_query(" 1 A St\t") == "1 a st" + + +def test_normalize_query_keeps_distinct_addresses_distinct(): + """Normalization does not merge addresses that differ in substance.""" + assert normalize_query("1 Main St") != normalize_query("1 Main Ave") + + +def test_duplicates_collapse_to_one_call_without_a_cache_file(): + """Repeated addresses in one run cost a single call even with no cache configured.""" + provider = _CountingProvider() + records = [ + _record(0, "1 Main St"), + _record(1, "1 MAIN ST"), + _record(2, "2 Oak St"), + _record(3, "1 Main St"), + ] + + results = provider.geocode(records) + + assert len(provider.fetched) == 2 + assert len(results) == 4 + assert results[0].result_address == results[1].result_address == results[3].result_address + assert results[2].result_address == "2 Oak St, Town, CA" + + +def test_cache_hit_across_runs_skips_the_api(tmp_path): + """A second run over an overlapping list only calls the API for the new addresses.""" + path = str(tmp_path / "cache.sqlite") + + with Cache(path) as cache: + first = _CountingProvider(cache) + first.geocode([_record(0, "1 Main St"), _record(1, "2 Oak St")]) + assert len(first.fetched) == 2 + + with Cache(path) as cache: + second = _CountingProvider(cache) + results = second.geocode([_record(0, "1 Main St"), _record(1, "3 Elm St")]) + + assert second.fetched == ["3 Elm St, Town, CA"] + assert results[0].result_address == "1 Main St, Town, CA" + assert results[1].result_address == "3 Elm St, Town, CA" + + +def test_cache_is_scoped_per_api(tmp_path): + """One provider's stored response is never served to another provider.""" + path = str(tmp_path / "cache.sqlite") + + with Cache(path) as cache: + _CountingProvider(cache).geocode([_record(0, "1 Main St")]) + + with Cache(path) as cache: + other = _CountingProvider(cache) + other.name = "different" + other.geocode([_record(0, "1 Main St")]) + + assert other.fetched == ["1 Main St, Town, CA"] + + +def test_read_only_cache_is_used_but_not_written(tmp_path): + """A --cacheRead file answers lookups and is left untouched by the run.""" + shared = tmp_path / "shared.sqlite" + with Cache(str(shared)) as cache: + _CountingProvider(cache).geocode([_record(0, "1 Main St")]) + + before = shared.read_bytes() + + with Cache(str(tmp_path / "own.sqlite"), [str(shared)]) as cache: + provider = _CountingProvider(cache) + provider.geocode([_record(0, "1 Main St"), _record(1, "2 Oak St")]) + + assert provider.fetched == ["2 Oak St, Town, CA"] + assert shared.read_bytes() == before + + +def test_writable_cache_wins_over_read_only(tmp_path): + """The writable cache answers first when both files hold the same key.""" + stale = str(tmp_path / "stale.sqlite") + fresh = str(tmp_path / "fresh.sqlite") + key = normalize_query("1 Main St, Town, CA") + + with Cache(stale) as cache: + cache.store("counting", key, "1 Main St, Town, CA", {"echo": "from stale"}) + with Cache(fresh) as cache: + cache.store("counting", key, "1 Main St, Town, CA", {"echo": "from fresh"}) + + with Cache(fresh, [stale]) as cache: + result = _CountingProvider(cache).geocode([_record(0, "1 Main St")])[0] + + assert result.result_address == "from fresh" + + +def test_store_commits_in_batches_for_crash_safety(tmp_path): + """Responses reach disk before the run ends so a crash keeps completed calls.""" + path = str(tmp_path / "cache.sqlite") + cache = Cache(path) + for index in range(Cache.COMMIT_INTERVAL): + cache.store("counting", f"key {index}", f"key {index}", {"echo": index}) + + committed = sqlite3.connect(path).execute("SELECT COUNT(*) FROM cache").fetchone()[0] + cache.close() + + assert committed == Cache.COMMIT_INTERVAL + + +def test_stored_row_keeps_the_original_query(tmp_path): + """The un-normalized query is retained alongside its key for manual review.""" + path = str(tmp_path / "cache.sqlite") + with Cache(path) as cache: + _CountingProvider(cache).geocode([_record(0, "1 MAIN St")]) + + row = sqlite3.connect(path).execute("SELECT query, original_query FROM cache").fetchone() + assert row[0] == "1 main st, town, ca" + assert row[1] == "1 MAIN St, Town, CA" + + +def test_read_only_cache_rejects_a_foreign_database(tmp_path): + """A SQLite file without a cache table is refused instead of failing later.""" + path = str(tmp_path / "other.sqlite") + connection = sqlite3.connect(path) + connection.execute("CREATE TABLE unrelated (id INTEGER)") + connection.commit() + connection.close() + + with pytest.raises(ValueError): + Cache(None, [path]) + + +def test_read_only_cache_rejects_a_non_database(tmp_path): + """A file that is not a SQLite database is refused with a clear error.""" + path = tmp_path / "notes.txt" + path.write_text("not a database") + + with pytest.raises(ValueError): + Cache(None, [str(path)]) + + +def test_writable_cache_rejects_a_non_database(tmp_path): + """Pointing --cache at an existing non-database file is refused, not overwritten.""" + path = tmp_path / "addresses.xlsx" + path.write_text("not a database") + + with pytest.raises(ValueError): + Cache(str(path)) + + assert path.read_text() == "not a database" + + +def test_missing_files_reports_only_absent_paths(tmp_path): + """Existing paths are dropped and absent ones are reported in order.""" + present = tmp_path / "here.sqlite" + present.write_text("") + absent = str(tmp_path / "gone.sqlite") + + assert missing_files([str(present), absent]) == [absent] diff --git a/test/census_test.py b/test/census_test.py index 2d15142..926d02f 100644 --- a/test/census_test.py +++ b/test/census_test.py @@ -162,3 +162,51 @@ def fake_post(*_args, **_kwargs): ] with pytest.raises(ValueError): census.CensusProvider().geocode(records) + + +def test_census_cache_key_ignores_country_and_pins_dataset(): + """The key covers the posted components and the dataset, but not the unsent country.""" + provider = census.CensusProvider() + usa = SourceRecord(internal_key=0, address="1 Main St", city="Town", stateprov="CA", postalcode="90210", country="US") + blank = SourceRecord(internal_key=1, address="1 Main St", city="Town", stateprov="CA", postalcode="90210") + + assert provider.cache_key(usa) == provider.cache_key(blank) + assert census.CensusProvider.BENCHMARK in provider.cache_key(usa) + assert census.CensusProvider.VINTAGE in provider.cache_key(usa) + + +def test_census_posts_each_distinct_address_once(monkeypatch): + """Repeated addresses are collapsed so the posted CSV carries one row each.""" + posted = [] + + def fake_post(_url, files=None, **_kwargs): + posted.append(files["addressFile"][1]) + keys = [line.split(",")[0] for line in files["addressFile"][1].splitlines() if line] + return _FakeResponse("".join(f'"{key}","q","No_Match"\r\n' for key in keys)) + + monkeypatch.setattr(census.requests, "post", fake_post) + + records = [ + SourceRecord(internal_key=0, address="1 Main St", city="Town", stateprov="CA"), + SourceRecord(internal_key=1, address="1 MAIN ST", city="Town", stateprov="CA"), + SourceRecord(internal_key=2, address="2 Oak St", city="Town", stateprov="CA"), + ] + results = census.CensusProvider().geocode(records) + + assert len(posted) == 1 + assert len([line for line in posted[0].splitlines() if line]) == 2 + assert len(results) == 3 + + +def test_census_raw_drops_the_run_local_id(monkeypatch): + """The echoed internal key is not kept, since it means nothing outside its run.""" + + def fake_post(*_args, **_kwargs): + return _FakeResponse('"0","1 Main St, Town, CA","Match","Exact","1 MAIN ST, TOWN, CA, 90210","-118.0,34.0","1","L","06","037","1","1"\r\n') + + monkeypatch.setattr(census.requests, "post", fake_post) + + result = census.CensusProvider().geocode([SourceRecord(internal_key=0, address="1 Main St", city="Town", stateprov="CA")])[0] + + assert "id" not in result.raw + assert result.raw["match_status"] == "Match" diff --git a/test/geocoder_test.py b/test/geocoder_test.py index 101b0e4..63c47ac 100644 --- a/test/geocoder_test.py +++ b/test/geocoder_test.py @@ -26,26 +26,33 @@ class MockProvider(Provider): name = "mock" requires_key = False - def geocode(self, records): - """Returns a fixed result per record, echoing the source address fields.""" - results = [] + def _fetch(self, records): + """Yields a canned response per record, echoing the source address fields.""" for record in records: - results.append( - GeocodeResult( - result_address=record.address, - result_city=record.city, - result_stateprov=record.stateprov, - result_country=record.country, - latitude="40.0", - longitude="-75.0", - match_type="exact", - accuracy=100, - location_type="rooftop", - match_notes="", - raw={"key": record.internal_key, "q": record.address_string()}, - ) - ) - return results + yield record, { + "key": record.internal_key, + "q": record.address_string(), + "address": record.address, + "city": record.city, + "stateprov": record.stateprov, + "country": record.country, + } + + def parse(self, raw): + """Rebuilds the echoed result from a raw response.""" + return GeocodeResult( + result_address=raw["address"], + result_city=raw["city"], + result_stateprov=raw["stateprov"], + result_country=raw["country"], + latitude="40.0", + longitude="-75.0", + match_type="exact", + accuracy=100, + location_type="rooftop", + match_notes="", + raw=raw, + ) def _make_workbook(path, sheets): @@ -299,6 +306,59 @@ def test_write_output_sheet_length_mismatch_raises(): write_output_sheet(openpyxl.Workbook(), "S", records, [], "mock", False) +def test_main_cache_flag_persists_across_runs(tmp_path): + """--cache creates the file, and a rerun of the same input calls nothing.""" + cache_path = tmp_path / "cache.sqlite" + infile = tmp_path / "in.xlsx" + _make_workbook(infile, {"S": [["Address", "City", "State"], ["1 A St", "Town", "CA"], ["1 A St", "Town", "CA"]]}) + + calls = [] + + class _CountingMock(MockProvider): + """Records every record handed to the API so calls can be counted.""" + + def _fetch(self, records): + """Counts the fetched records before delegating to the mock response.""" + calls.extend(record.address_string() for record in records) + yield from super()._fetch(records) + + api.PROVIDERS["mock"] = _CountingMock + try: + main([str(infile), str(tmp_path / "a.xlsx"), "--api", "mock", "--cache", str(cache_path)]) + assert calls == ["1 A St, Town, CA"] + + main([str(infile), str(tmp_path / "b.xlsx"), "--api", "mock", "--cache", str(cache_path)]) + finally: + del api.PROVIDERS["mock"] + + assert calls == ["1 A St, Town, CA"] + assert cache_path.is_file() + + +def test_main_missing_read_cache_exits(tmp_path): + """A --cacheRead file that does not exist exits before any provider work.""" + infile = tmp_path / "in.xlsx" + _make_workbook(infile, {"S": [["Address", "City", "State"], ["1 A St", "Town", "CA"]]}) + + with pytest.raises(SystemExit): + main([str(infile), str(tmp_path / "out.xlsx"), "--api", "mock", "--cacheRead", str(tmp_path / "gone.sqlite")]) + + +def test_main_unusable_read_cache_exits(tmp_path): + """A --cacheRead file that is not a geocoder cache exits with a message.""" + infile = tmp_path / "in.xlsx" + _make_workbook(infile, {"S": [["Address", "City", "State"], ["1 A St", "Town", "CA"]]}) + foreign = tmp_path / "foreign.sqlite" + foreign.write_text("not a database") + + api.PROVIDERS["mock"] = MockProvider + try: + with pytest.raises(SystemExit): + main([str(infile), str(tmp_path / "out.xlsx"), "--api", "mock", "--cacheRead", str(foreign)]) + finally: + del api.PROVIDERS["mock"] + + def test_main_runs_census_provider(tmp_path, monkeypatch): """--api census runs end to end and writes the census result columns.""" infile = tmp_path / "in.xlsx" diff --git a/test/google_test.py b/test/google_test.py index c05513f..d0a1946 100644 --- a/test/google_test.py +++ b/test/google_test.py @@ -12,6 +12,7 @@ from src import google from src.api import PROVIDERS, SourceRecord +from src.cache import Cache, normalize_query class _FakeResponse: @@ -160,3 +161,48 @@ def test_google_raises_on_error_status(monkeypatch): with pytest.raises(ValueError): google.GoogleProvider("key").geocode([SourceRecord(internal_key=0, address="1 Main St")]) + + +def test_google_raw_is_the_whole_response(monkeypatch): + """The stored raw value is the full envelope, not just the match read from it.""" + _patch_response(monkeypatch, _result("ROOFTOP", ROOFTOP_COMPONENTS)) + + result = google.GoogleProvider("key").geocode([SourceRecord(internal_key=0, address="1600 Pennsylvania Ave NW")])[0] + + assert result.raw["status"] == "OK" + assert result.raw["results"][0]["place_id"] == "PLACE" + + +def test_google_requests_each_distinct_address_once(monkeypatch): + """Records sharing an address cost one request and all receive the result.""" + requested = [] + + def fake_get(_url, params=None, **_kwargs): + requested.append(params["address"]) + return _FakeResponse(_result("ROOFTOP", ROOFTOP_COMPONENTS)) + + monkeypatch.setattr(google.requests, "get", fake_get) + + records = [ + SourceRecord(internal_key=0, address="1 Main St", city="Town", stateprov="CA"), + SourceRecord(internal_key=1, address="1 MAIN ST", city="Town", stateprov="CA"), + SourceRecord(internal_key=2, address="2 Oak St", city="Town", stateprov="CA"), + ] + results = google.GoogleProvider("key").geocode(records) + + assert len(requested) == 2 + assert len(results) == 3 + assert all(result.accuracy == 100 for result in results) + + +def test_google_error_status_is_never_cached(monkeypatch, tmp_path): + """A failed request leaves nothing behind, so a later run retries it.""" + path = str(tmp_path / "cache.sqlite") + _patch_response(monkeypatch, {"status": "OVER_QUERY_LIMIT", "error_message": "quota exceeded"}) + + with Cache(path) as cache: + with pytest.raises(ValueError): + google.GoogleProvider("key", cache).geocode([SourceRecord(internal_key=0, address="1 Main St")]) + + with Cache(path) as cache: + assert not cache.lookup("google", [normalize_query("1 Main St")]) From 36815c9084ee7fb21ba4b12bb93942a0729fd6ce Mon Sep 17 00:00:00 2001 From: Zachary Schwartz <112527479+ZachSchwartz@users.noreply.github.com> Date: Tue, 1 Sep 2026 16:45:15 -0500 Subject: [PATCH 2/7] making nocache optional --- src/api.py | 13 ++++++++++-- src/cache.py | 2 ++ src/census.py | 20 +++++++++++++++++++ src/geocoder.py | 18 +++++++++++++---- src/google.py | 10 ++++++++++ test/geocoder_test.py | 46 ++++++++++++++++++++++++++++++++++++++++--- 6 files changed, 100 insertions(+), 9 deletions(-) diff --git a/src/api.py b/src/api.py index 3712fc9..1b27168 100644 --- a/src/api.py +++ b/src/api.py @@ -204,6 +204,16 @@ def parse(self, raw: Dict[str, Any]) -> GeocodeResult: stays live: a stored response is always scored by the current rules rather than the ones in force when it was fetched. This must be a pure mapping over the response and must not call the API. + + Parameters + ---------- + raw : Dict[str, Any] + One response as the provider returned it. + + Return + ---------- + GeocodeResult + The normalized result that response describes. """ def geocode(self, records: List[SourceRecord]) -> List[GeocodeResult]: @@ -225,8 +235,7 @@ def geocode(self, records: List[SourceRecord]) -> List[GeocodeResult]: List[GeocodeResult] One result per input record, aligned by position. """ - queries = [self.cache_key(record) for record in records] - keys = [normalize_query(query) for query in queries] + keys = [normalize_query(self.cache_key(record)) for record in records] responses = self.cache.lookup(self.name, keys) pending: Dict[str, SourceRecord] = {} diff --git a/src/cache.py b/src/cache.py index 62b8cf9..9cafce5 100644 --- a/src/cache.py +++ b/src/cache.py @@ -14,6 +14,8 @@ from pathlib import Path from typing import Any, Dict, Iterable, Iterator, List, Optional, Sequence +DEFAULT_CACHE_FILE = "geocoder-cache.sqlite" + SCHEMA = """ CREATE TABLE IF NOT EXISTS cache ( api TEXT NOT NULL, diff --git a/src/census.py b/src/census.py index bada20e..a39abee 100644 --- a/src/census.py +++ b/src/census.py @@ -63,6 +63,16 @@ def cache_key(self, record: SourceRecord) -> str: benchmark and vintage are included because they select which dataset answers the query, and moving off Census2020 would make stored responses answers to a different question. + + Parameters + ---------- + record : SourceRecord + The record whose query is being composed. + + Return + ---------- + str + The query text identifying this record's response. """ parts = [record.address, record.city, record.stateprov, record.postalcode, self.BENCHMARK, self.VINTAGE] return "|".join(parts) @@ -154,6 +164,16 @@ def parse(self, raw: Dict[str, Any]) -> GeocodeResult: Accuracy is graded from the populated result fields rather than the Census match type, so it reflects how specific the returned location is. + + Parameters + ---------- + raw : Dict[str, Any] + One named Census response row. + + Return + ---------- + GeocodeResult + The normalized, graded result that row describes. """ result = self._build_result(raw) result.accuracy = grade_accuracy(result, self.MAX_ACCURACY) diff --git a/src/geocoder.py b/src/geocoder.py index 395fbfb..da80585 100644 --- a/src/geocoder.py +++ b/src/geocoder.py @@ -23,7 +23,7 @@ SourceRecord, resolve_api_key, ) -from .cache import Cache, missing_files +from .cache import DEFAULT_CACHE_FILE, Cache, missing_files CANONICAL_FIELDS = [ "ID", @@ -333,11 +333,18 @@ def parse_args(argv: Optional[List[str]] = None): action="store_true", help="use the worksheet name as COUNTRY when a row's country is blank", ) - parser.add_argument( + cache_group = parser.add_mutually_exclusive_group() + cache_group.add_argument( "--cache", default=None, metavar="FILE", - help="SQLite cache of prior API responses, read and written (created if absent)", + help=f"SQLite cache of prior API responses, read and written " f"(default: {DEFAULT_CACHE_FILE} in the working directory)", + ) + cache_group.add_argument( + "--noCache", + dest="no_cache", + action="store_true", + help="do not read or write a cache file; repeated addresses within the run are still collapsed", ) parser.add_argument( "--cacheRead", @@ -394,12 +401,15 @@ def main(argv: Optional[List[str]] = None) -> None: if not api_key: raise SystemExit(f"error: api '{args.api}' requires an API key " f"(pass --apiKey or set {KEY_ENV_VARS[args.api]})") + cache_path = None if args.no_cache else args.cache or DEFAULT_CACHE_FILE try: - cache = Cache(args.cache, args.cache_read) + cache = Cache(cache_path, args.cache_read) except ValueError as error: raise SystemExit(f"error: {error}") from error with cache: + if cache_path: + print(f"using cache {os.path.abspath(cache_path)}") process_workbook( args.infile, args.outfile, diff --git a/src/google.py b/src/google.py index ea77e58..9e30cf0 100644 --- a/src/google.py +++ b/src/google.py @@ -102,6 +102,16 @@ def parse(self, raw: Dict[str, Any]) -> GeocodeResult: The graded accuracy is capped at the tier implied by ``location_type`` so an interpolated or centroid match cannot report rooftop precision on the strength of the echoed address fields. + + Parameters + ---------- + raw : Dict[str, Any] + One Google response envelope. + + Return + ---------- + GeocodeResult + The normalized, graded result that response describes. """ if raw.get("status") != "OK": return GeocodeResult(match_notes="No match", raw=raw) diff --git a/test/geocoder_test.py b/test/geocoder_test.py index 63c47ac..3b38c34 100644 --- a/test/geocoder_test.py +++ b/test/geocoder_test.py @@ -17,6 +17,7 @@ from src import census from src import geocoder from src.api import GeocodeResult, Provider, SourceRecord +from src.cache import DEFAULT_CACHE_FILE from src.geocoder import detect_columns, main, process_workbook, write_output_sheet @@ -292,13 +293,52 @@ def test_main_runs_registered_provider(tmp_path): api.PROVIDERS["mock"] = MockProvider try: - main([str(infile), str(outfile), "--api", "mock"]) + main([str(infile), str(outfile), "--api", "mock", "--noCache"]) finally: del api.PROVIDERS["mock"] assert outfile.exists() +def test_main_defaults_to_a_cache_in_the_working_directory(tmp_path, monkeypatch): + """With no cache flag the default file is created beside the working directory.""" + infile = tmp_path / "in.xlsx" + _make_workbook(infile, {"S": [["Address", "City", "State"], ["1 A St", "Town", "CA"]]}) + monkeypatch.chdir(tmp_path) + + api.PROVIDERS["mock"] = MockProvider + try: + main([str(infile), "out.xlsx", "--api", "mock"]) + finally: + del api.PROVIDERS["mock"] + + assert (tmp_path / DEFAULT_CACHE_FILE).is_file() + + +def test_main_no_cache_writes_nothing(tmp_path, monkeypatch): + """--noCache leaves no cache file behind, including the default one.""" + infile = tmp_path / "in.xlsx" + _make_workbook(infile, {"S": [["Address", "City", "State"], ["1 A St", "Town", "CA"]]}) + monkeypatch.chdir(tmp_path) + + api.PROVIDERS["mock"] = MockProvider + try: + main([str(infile), "out.xlsx", "--api", "mock", "--noCache"]) + finally: + del api.PROVIDERS["mock"] + + assert not (tmp_path / DEFAULT_CACHE_FILE).exists() + + +def test_main_rejects_cache_and_no_cache_together(tmp_path): + """Naming a cache file while disabling the cache is contradictory and exits.""" + infile = tmp_path / "in.xlsx" + _make_workbook(infile, {"S": [["Address", "City", "State"], ["1 A St", "Town", "CA"]]}) + + with pytest.raises(SystemExit): + main([str(infile), str(tmp_path / "out.xlsx"), "--api", "mock", "--cache", str(tmp_path / "c.sqlite"), "--noCache"]) + + def test_write_output_sheet_length_mismatch_raises(): """A provider returning the wrong number of results is rejected.""" records = [SourceRecord(internal_key=0, address="1 A St")] @@ -354,7 +394,7 @@ def test_main_unusable_read_cache_exits(tmp_path): api.PROVIDERS["mock"] = MockProvider try: with pytest.raises(SystemExit): - main([str(infile), str(tmp_path / "out.xlsx"), "--api", "mock", "--cacheRead", str(foreign)]) + main([str(infile), str(tmp_path / "out.xlsx"), "--api", "mock", "--noCache", "--cacheRead", str(foreign)]) finally: del api.PROVIDERS["mock"] @@ -378,7 +418,7 @@ def fake_post(*_args, **_kwargs): monkeypatch.setattr(census.requests, "post", fake_post) - main([str(infile), str(outfile), "--api", "census"]) + main([str(infile), str(outfile), "--api", "census", "--noCache"]) sheet = openpyxl.load_workbook(outfile)["S"] header = [cell.value for cell in sheet[1]] From 58b7e5b3c156149c4541056eba710f661429cb05 Mon Sep 17 00:00:00 2001 From: Zachary Schwartz <112527479+ZachSchwartz@users.noreply.github.com> Date: Wed, 2 Sep 2026 10:08:08 -0500 Subject: [PATCH 3/7] trimming docstrings --- src/api.py | 32 ++++++++++---------------------- src/cache.py | 32 ++++++++------------------------ src/census.py | 15 ++++----------- 3 files changed, 22 insertions(+), 57 deletions(-) diff --git a/src/api.py b/src/api.py index 1b27168..673c65a 100644 --- a/src/api.py +++ b/src/api.py @@ -145,10 +145,8 @@ class Provider(ABC): """ Base class for geocoding providers; subclasses self-register via @register. - Subclasses supply three pieces and inherit the rest: ``_fetch`` calls the API, - ``parse`` turns one raw response into a result, and ``cache_key`` names the - query a record resolves to. Deduplication and caching are handled once here so - no provider has to repeat them. + Subclasses supply ``_fetch``, ``parse``, and ``cache_key``; deduplication and + caching are handled once here so no provider repeats them. """ name: str = "" @@ -165,11 +163,8 @@ def cache_key(self, record: SourceRecord) -> str: """ Returns the query text this provider would send for the given record. - Two records with the same key are answered by a single API call, so the - key must cover everything that changes the response — the address - components a provider actually submits plus any pinned request parameters - that steer the result. Providers whose query is not the full address - override this. + Two records with the same key are answered by a single API call, so the key + must cover everything that changes the response. Parameters ---------- @@ -188,11 +183,8 @@ def _fetch(self, records: List[SourceRecord]) -> Iterator[Tuple[SourceRecord, Di """ Yields each record paired with its raw provider response, as replies arrive. - Yielding rather than returning lets the caller cache responses at whatever - granularity the provider produces them, so an interrupted run keeps the - calls it already paid for. Only responses that have been validated against - the provider's expected shape may be yielded, because everything yielded is - cached; a malformed or failed response must raise instead. + Everything yielded is cached, so only responses validated against the + provider's expected shape may be yielded; a failed response must raise. """ @abstractmethod @@ -200,10 +192,8 @@ def parse(self, raw: Dict[str, Any]) -> GeocodeResult: """ Converts one raw provider response into a normalized, graded GeocodeResult. - Cached and freshly fetched responses both come through here, so grading - stays live: a stored response is always scored by the current rules rather - than the ones in force when it was fetched. This must be a pure mapping - over the response and must not call the API. + Cached and freshly fetched responses both come through here, so a stored + response is always scored by the current grading rules. Parameters ---------- @@ -220,10 +210,8 @@ def geocode(self, records: List[SourceRecord]) -> List[GeocodeResult]: """ Returns one GeocodeResult per input record, in the same order. - Records sharing a query are collapsed to a single API call, and any query - already in the cache skips the API entirely, so a list of 5000 addresses - costs one call per distinct address never seen before. Deduplication - applies even with no cache file configured. + Records sharing a query are collapsed to a single API call and cached + queries skip the API entirely, with or without a cache file configured. Parameters ---------- diff --git a/src/cache.py b/src/cache.py index 9cafce5..d6daf5b 100644 --- a/src/cache.py +++ b/src/cache.py @@ -32,10 +32,9 @@ def normalize_query(query: str) -> str: """ Canonicalizes a provider query into the key its response is stored under. - Geocoders ignore letter case and runs of whitespace, so folding both lets - "123 Main St" and "123 MAIN ST" share a single cached response rather than - costing two identical API calls. Nothing else is altered: punctuation and - abbreviations can change what a geocoder returns, so they are left alone. + Letter case and runs of whitespace are folded so equivalent queries share one + cached response; nothing else is altered, since punctuation and abbreviations + can change what a geocoder returns. Parameters ---------- @@ -54,20 +53,8 @@ class Cache: """ A SQLite-backed store of raw provider responses keyed by the query that produced them. - The cache is never a source of truth. It holds only what a provider actually - returned, so deleting any or all of its files changes nothing about the output - beyond how many API calls a run costs. Results are parsed from the stored - response on every run, which keeps a cached row scored by the current grading - rules rather than the ones in force when it was fetched. - - One table serves every provider, so adding a provider needs no schema change - and consolidating two files is a single statement:: - - ATTACH 'other.sqlite' AS other; - INSERT OR IGNORE INTO cache SELECT * FROM other.cache; - - A cache with no writable path and no readers is a working no-op, which lets - providers use one unconditionally. + It is never a source of truth: deleting a cache file changes nothing but how + many API calls a run costs, and one table serves every provider. """ COMMIT_INTERVAL = 250 @@ -149,10 +136,8 @@ def lookup(self, api: str, keys: Iterable[str]) -> Dict[str, Any]: """ Fetches the stored responses for the given keys, writable cache first. - Each key is looked up in the writable cache and then in each read-only - cache, and the first file holding it wins. Keys with no stored response - are absent from the result. Lookups are chunked so a run of any size stays - within the SQLite limit on bound parameters. + The first file holding a key wins, and keys with no stored response are + absent from the result. Parameters ---------- @@ -187,8 +172,7 @@ def store(self, api: str, key: str, query: str, response: Any) -> None: Records one provider response, committing once a batch has accumulated. Committing as the run proceeds means a crash costs only the calls made - since the last commit rather than every call made so far. Nothing is - written when no writable cache is configured. + since the last commit. Parameters ---------- diff --git a/src/census.py b/src/census.py index a39abee..ccc0a7d 100644 --- a/src/census.py +++ b/src/census.py @@ -58,11 +58,8 @@ def cache_key(self, record: SourceRecord) -> str: """ Builds the key from the components Census is sent, plus the pinned dataset. - Country is excluded because the addressbatch CSV has no country column, so - two rows differing only in country resolve to the same request. The - benchmark and vintage are included because they select which dataset - answers the query, and moving off Census2020 would make stored responses - answers to a different question. + Country is excluded because the addressbatch CSV has no country column, + while the benchmark and vintage select which dataset answers the query. Parameters ---------- @@ -143,12 +140,8 @@ def _build_raw(self, row: List[str]) -> Dict[str, Any]: """ Names the positional fields of one response row, validating a match first. - Reading the headerless response by position is only safe while the pinned - layout holds, so a match row is checked here, where it arrives from the - network, and never again. Everything downstream reads the named fields. - - The echoed id is dropped: it is this run's internal key, which carries no - meaning once the response outlives the run that fetched it. + The pinned layout is checked here, where the row arrives from the network, + and the echoed id is dropped as it is only meaningful within this run. """ status = row[2] if len(row) > 2 else "No_Match" if status == "Match": From e913149c8afb7f2a8e8354994d368754ca57ff5c Mon Sep 17 00:00:00 2001 From: Zachary Schwartz <112527479+ZachSchwartz@users.noreply.github.com> Date: Wed, 2 Sep 2026 13:09:13 -0500 Subject: [PATCH 4/7] Update api.py --- src/api.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/api.py b/src/api.py index 673c65a..ff4d2e5 100644 --- a/src/api.py +++ b/src/api.py @@ -231,6 +231,9 @@ def geocode(self, records: List[SourceRecord]) -> List[GeocodeResult]: if key not in responses and key not in pending: pending[key] = record + if records: + print(f"{len(records)} rows, {len(responses) + len(pending)} distinct queries: {len(responses)} from cache, {len(pending)} to fetch") + if pending: for record, raw in self._fetch(list(pending.values())): query = self.cache_key(record) From b6960a0c7586ada8c2652548835ac4fd75617a39 Mon Sep 17 00:00:00 2001 From: Zachary Schwartz <112527479+ZachSchwartz@users.noreply.github.com> Date: Wed, 16 Sep 2026 10:39:05 -0500 Subject: [PATCH 5/7] fix --- src/cache.py | 19 +++++++++--- src/census.py | 7 ++++- src/geocodio.py | 67 ++++++++++++++++++++++++++----------------- test/cache_test.py | 12 ++++++++ test/census_test.py | 9 ++++++ test/geocodio_test.py | 23 +++++++++++++++ test/google_test.py | 2 +- 7 files changed, 106 insertions(+), 33 deletions(-) diff --git a/src/cache.py b/src/cache.py index d6daf5b..9e6d7d1 100644 --- a/src/cache.py +++ b/src/cache.py @@ -16,6 +16,8 @@ DEFAULT_CACHE_FILE = "geocoder-cache.sqlite" +REQUIRED_COLUMNS = ("api", "query", "response") + SCHEMA = """ CREATE TABLE IF NOT EXISTS cache ( api TEXT NOT NULL, @@ -97,7 +99,11 @@ def _open_writer(path: str) -> sqlite3.Connection: @staticmethod def _open_reader(path: str) -> sqlite3.Connection: """ - Opens a cache file read-only and confirms it carries the cache table. + Opens a cache file read-only and confirms its cache table is usable. + + The columns lookups read are checked as well as the table itself, so a + foreign database that happens to hold a differently shaped cache table is + refused here rather than failing mid-run on the first lookup. Parameters ---------- @@ -112,18 +118,23 @@ def _open_reader(path: str) -> sqlite3.Connection: Raises ---------- ValueError - If the file is not a SQLite database or holds no cache table. + If the file is not a SQLite database or holds no usable cache table. """ connection = sqlite3.connect(f"{Path(path).resolve().as_uri()}?mode=ro", uri=True) try: - found = connection.execute("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'cache'").fetchone() + columns = {row[1] for row in connection.execute("PRAGMA table_info(cache)")} except sqlite3.DatabaseError as error: connection.close() raise ValueError(f"'{path}' is not a readable SQLite database: {error}") from error - if found is None: + if not columns: connection.close() raise ValueError(f"'{path}' is not a geocoder cache (it has no 'cache' table)") + + missing = [column for column in REQUIRED_COLUMNS if column not in columns] + if missing: + connection.close() + raise ValueError(f"'{path}' is not a geocoder cache (its 'cache' table is missing {', '.join(missing)})") return connection def _connections(self) -> Iterator[sqlite3.Connection]: diff --git a/src/census.py b/src/census.py index ccc0a7d..29501ba 100644 --- a/src/census.py +++ b/src/census.py @@ -10,6 +10,7 @@ import csv import io +import json from typing import Any, Dict, Iterator, List, Tuple import requests @@ -61,6 +62,10 @@ def cache_key(self, record: SourceRecord) -> str: Country is excluded because the addressbatch CSV has no country column, while the benchmark and vintage select which dataset answers the query. + The components are JSON-encoded rather than joined on a separator, so a + row whose fields happen to contain that separator cannot produce the same + key as a different row and be answered with its response. + Parameters ---------- record : SourceRecord @@ -72,7 +77,7 @@ def cache_key(self, record: SourceRecord) -> str: The query text identifying this record's response. """ parts = [record.address, record.city, record.stateprov, record.postalcode, self.BENCHMARK, self.VINTAGE] - return "|".join(parts) + return json.dumps(parts, ensure_ascii=False) def _fetch(self, records: List[SourceRecord]) -> Iterator[Tuple[SourceRecord, Dict[str, Any]]]: """ diff --git a/src/geocodio.py b/src/geocodio.py index 93a21cb..db0fffe 100644 --- a/src/geocodio.py +++ b/src/geocodio.py @@ -8,7 +8,7 @@ Copyright (c) 2026 Pangaea Information Technologies, Ltd. """ -from typing import Dict, List +from typing import Any, Dict, Iterator, List, Tuple import requests @@ -60,9 +60,12 @@ class GeocodioProvider(Provider): EXACT_TYPES = {"rooftop", "point"} - def geocode(self, records: List[SourceRecord]) -> List[GeocodeResult]: + def _fetch(self, records: List[SourceRecord]) -> Iterator[Tuple[SourceRecord, Dict[str, Any]]]: """ - Geocodes records in batches and returns one result per record, in order. + Posts records in batches, yielding each record with its response entry. + + Yielding per batch rather than per run means an interrupted job keeps + every batch that already came back. Parameters ---------- @@ -71,22 +74,18 @@ def geocode(self, records: List[SourceRecord]) -> List[GeocodeResult]: Return ---------- - List[GeocodeResult] - One result per input record, aligned by position. + Iterator[Tuple[SourceRecord, Dict[str, Any]]] + Each record paired with its raw Geocodio response entry. """ - results_by_key: Dict[int, GeocodeResult] = {} for start in range(0, len(records), self.BATCH_SIZE): - self._geocode_batch(records[start : start + self.BATCH_SIZE], results_by_key) - - return [results_by_key.get(record.internal_key, GeocodeResult(match_notes="No match")) for record in records] + yield from self._fetch_batch(records[start : start + self.BATCH_SIZE]) - def _geocode_batch(self, batch: List[SourceRecord], results_by_key: Dict[int, GeocodeResult]) -> None: - """Posts one batch, verifies its entry count, and stores each result by internal key.""" + def _fetch_batch(self, batch: List[SourceRecord]) -> Iterator[Tuple[SourceRecord, Dict[str, Any]]]: + """Posts one batch, verifies its entry count, and matches entries back by position.""" entries = self._post_batch(batch) if len(entries) != len(batch): raise ValueError(f"Geocodio returned {len(entries)} entries for {len(batch)} submitted records; the batch response may have changed") - for record, entry in zip(batch, entries): - results_by_key[record.internal_key] = self._parse_entry(entry) + yield from zip(batch, entries) def _post_batch(self, batch: List[SourceRecord]) -> List[Dict]: """Posts one batch through the retrying request helper and returns the ordered result entries.""" @@ -101,23 +100,30 @@ def _post_batch(self, batch: List[SourceRecord]) -> List[Dict]: ) return response.json().get("results", []) - def _parse_entry(self, entry: Dict) -> GeocodeResult: - """Grades the best candidate for one input, or returns a no-match when none was found.""" - matches = entry.get("response", {}).get("results", []) - if not matches: - return GeocodeResult(match_notes="No match", raw=entry) - return self._parse_result(matches[0]) - - def _parse_result(self, match: Dict) -> GeocodeResult: + def parse(self, raw: Dict[str, Any]) -> GeocodeResult: """ - Converts one Geocodio candidate into a normalized GeocodeResult. + Converts one Geocodio response entry into a normalized GeocodeResult. The graded accuracy is capped at the tier implied by ``accuracy_type`` so an interpolated or centroid match cannot report rooftop precision on the strength of the echoed address fields. + + Parameters + ---------- + raw : Dict[str, Any] + One Geocodio response entry. + + Return + ---------- + GeocodeResult + The normalized, graded result that entry describes. """ - components = match.get("address_components", {}) - result = self._build_result(match, components) + matches = raw.get("response", {}).get("results", []) + if not matches: + return GeocodeResult(match_notes="No match", raw=raw) + + components = matches[0].get("address_components", {}) + result = self._build_result(raw, components) result.accuracy = grade_accuracy(result, self._accuracy_cap(result, components)) return result @@ -134,8 +140,15 @@ def _accuracy_cap(self, result: GeocodeResult, components: Dict[str, str]) -> Ac return cap return min(cap, AccuracyLevel.STREET) - def _build_result(self, match: Dict, components: Dict[str, str]) -> GeocodeResult: - """Maps a Geocodio candidate to a GeocodeResult without scoring its accuracy.""" + def _build_result(self, raw: Dict[str, Any], components: Dict[str, str]) -> GeocodeResult: + """ + Maps the best Geocodio candidate to a GeocodeResult without scoring its accuracy. + + The whole response entry is kept as the raw value rather than the single + candidate it was read from, so what is cached and reported is what + Geocodio actually said. + """ + match = raw["response"]["results"][0] location = match.get("location", {}) accuracy_type = match.get("accuracy_type", "") @@ -149,7 +162,7 @@ def _build_result(self, match: Dict, components: Dict[str, str]) -> GeocodeResul longitude=str(location.get("lng", "")), match_type="exact" if accuracy_type in self.EXACT_TYPES else "non-exact", location_type=accuracy_type, - raw=match, + raw=raw, ) @staticmethod diff --git a/test/cache_test.py b/test/cache_test.py index d4caa14..0384dd1 100644 --- a/test/cache_test.py +++ b/test/cache_test.py @@ -172,6 +172,18 @@ def test_read_only_cache_rejects_a_foreign_database(tmp_path): Cache(None, [path]) +def test_read_only_cache_rejects_a_differently_shaped_cache_table(tmp_path): + """A 'cache' table without the columns lookups read is refused up front.""" + path = str(tmp_path / "other.sqlite") + connection = sqlite3.connect(path) + connection.execute("CREATE TABLE cache (key TEXT, value TEXT)") + connection.commit() + connection.close() + + with pytest.raises(ValueError): + Cache(None, [path]) + + def test_read_only_cache_rejects_a_non_database(tmp_path): """A file that is not a SQLite database is refused with a clear error.""" path = tmp_path / "notes.txt" diff --git a/test/census_test.py b/test/census_test.py index 926d02f..a506568 100644 --- a/test/census_test.py +++ b/test/census_test.py @@ -175,6 +175,15 @@ def test_census_cache_key_ignores_country_and_pins_dataset(): assert census.CensusProvider.VINTAGE in provider.cache_key(usa) +def test_census_cache_key_keeps_shifted_components_distinct(): + """Rows whose components differ only in where a field boundary falls get distinct keys.""" + provider = census.CensusProvider() + split = SourceRecord(internal_key=0, address="1 Main St|Town", city="CA") + shifted = SourceRecord(internal_key=1, address="1 Main St", city="Town|CA") + + assert provider.cache_key(split) != provider.cache_key(shifted) + + def test_census_posts_each_distinct_address_once(monkeypatch): """Repeated addresses are collapsed so the posted CSV carries one row each.""" posted = [] diff --git a/test/geocodio_test.py b/test/geocodio_test.py index d2af7de..1d35ee7 100644 --- a/test/geocodio_test.py +++ b/test/geocodio_test.py @@ -235,6 +235,29 @@ def test_geocodio_aligns_entries_by_position(monkeypatch): assert results[2].accuracy == 40 +def test_geocodio_posts_each_distinct_address_once(monkeypatch): + """Repeated addresses are collapsed so the posted batch carries one entry each.""" + posted = [] + + def fake_post(_url, json=None, **_kwargs): + posted.append(json) + return FakeResponse(_batch(*[[_candidate("rooftop", ROOFTOP_COMPONENTS)] for _ in json])) + + monkeypatch.setattr(geocodio.requests, "post", fake_post) + + records = [ + SourceRecord(internal_key=0, address="1 Main St", city="Town", stateprov="CA"), + SourceRecord(internal_key=1, address="1 MAIN ST", city="Town", stateprov="CA"), + SourceRecord(internal_key=2, address="2 Oak St", city="Town", stateprov="CA"), + ] + results = geocodio.GeocodioProvider("key").geocode(records) + + assert len(posted) == 1 + assert len(posted[0]) == 2 + assert len(results) == 3 + assert all(result.accuracy == 100 for result in results) + + def test_geocodio_raises_on_entry_count_mismatch(monkeypatch): """A response short of one entry per record fails loudly instead of misaligning.""" _patch_response(monkeypatch, _batch([_candidate("rooftop", ROOFTOP_COMPONENTS)])) diff --git a/test/google_test.py b/test/google_test.py index ff4e7fc..8aa44a1 100644 --- a/test/google_test.py +++ b/test/google_test.py @@ -249,7 +249,7 @@ def test_google_requests_each_distinct_address_once(monkeypatch): def fake_get(_url, params=None, **_kwargs): requested.append(params["address"]) - return _FakeResponse(_result("ROOFTOP", ROOFTOP_COMPONENTS)) + return FakeResponse(_result("ROOFTOP", ROOFTOP_COMPONENTS)) monkeypatch.setattr(google.requests, "get", fake_get) From 322b48b78facbcbfe9ae98092b041ea4fab423ca Mon Sep 17 00:00:00 2001 From: Zachary Schwartz <112527479+ZachSchwartz@users.noreply.github.com> Date: Mon, 21 Sep 2026 11:48:44 -0500 Subject: [PATCH 6/7] commit --- README.md | 9 ++++ src/cache.py | 119 +++++++++++++++++++++----------------------- src/census.py | 30 ++--------- test/cache_test.py | 113 +++++++++++++++++++++++++++-------------- test/census_test.py | 22 ++------ 5 files changed, 150 insertions(+), 143 deletions(-) diff --git a/README.md b/README.md index b70eb13..99cf52a 100644 --- a/README.md +++ b/README.md @@ -91,6 +91,15 @@ poetry run geocoder input.xlsx output.xlsx --api census --cacheRead archive.sqli poetry run geocoder input.xlsx output.xlsx --api census --noCache ``` +`--noCache` opens no cache file at all, so `--cache` and `--cacheRead` are +ignored when it is given; repeated addresses within a run still cost one call. +A `--cacheRead` file must hold entries for the provider being run, and is +rejected if it does not. + +Each provider's responses are stored in a table of their own, tagged with the +version of the API that produced them: entries written by an earlier version are +never served, and are replaced as their addresses are looked up again. + ## Providers | Provider | `--api` value | Coverage | API key | diff --git a/src/cache.py b/src/cache.py index 9e6d7d1..07afe4c 100644 --- a/src/cache.py +++ b/src/cache.py @@ -16,61 +16,54 @@ DEFAULT_CACHE_FILE = "geocoder-cache.sqlite" -REQUIRED_COLUMNS = ("api", "query", "response") +REQUIRED_COLUMNS = ("query", "version", "response") SCHEMA = """ -CREATE TABLE IF NOT EXISTS cache ( - api TEXT NOT NULL, - query TEXT NOT NULL, - original_query TEXT NOT NULL, +CREATE TABLE IF NOT EXISTS {table} ( + query TEXT PRIMARY KEY, + version TEXT NOT NULL, response TEXT NOT NULL, - fetched_at TEXT NOT NULL, - PRIMARY KEY (api, query) + fetched_at TEXT NOT NULL ) """ -def normalize_query(query: str) -> str: - """ - Canonicalizes a provider query into the key its response is stored under. - - Letter case and runs of whitespace are folded so equivalent queries share one - cached response; nothing else is altered, since punctuation and abbreviations - can change what a geocoder returns. - - Parameters - ---------- - query : str - The query text a provider would send for a record. - - Return - ---------- - str - The canonical key for that query. - """ - return " ".join(query.split()).casefold() - - class Cache: """ A SQLite-backed store of raw provider responses keyed by the query that produced them. - It is never a source of truth: deleting a cache file changes nothing but how - many API calls a run costs, and one table serves every provider. + The key is the address a run asks about, and nothing else: how the provider + was called is held apart from it, as the version tag every row carries. + Lookups are filtered to the version the run is calling now, so entries + written under an earlier one are never served, and re-fetching a query + overwrites its row rather than leaving an obsolete one behind for good. + + Each provider owns a table named for it, so a lookup reads only the entries + that could answer it and the tables of providers never called cost nothing. + + A cache is never a source of truth: deleting a cache file changes nothing + but how many API calls a run costs. """ COMMIT_INTERVAL = 250 LOOKUP_CHUNK = 500 - def __init__(self, path: Optional[str] = None, read_paths: Sequence[str] = ()): + def __init__(self, api: str, version: str, path: Optional[str] = None, read_paths: Sequence[str] = ()): + self._api = api + self._version = version self._writer = self._open_writer(path) if path else None self._readers = [self._open_reader(read_path) for read_path in read_paths] self._uncommitted = 0 - @staticmethod - def _open_writer(path: str) -> sqlite3.Connection: + @property + def _table(self) -> str: + """Returns this provider's table name, quoted for use in a statement.""" + escaped = self._api.replace('"', '""') + return f'"{escaped}"' + + def _open_writer(self, path: str) -> sqlite3.Connection: """ - Opens the read-write cache, creating the file and table when absent. + Opens the read-write cache, creating the file and this provider's table when absent. Parameters ---------- @@ -80,7 +73,7 @@ def _open_writer(path: str) -> sqlite3.Connection: Return ---------- sqlite3.Connection - A connection with the cache table in place. + A connection with this provider's table in place. Raises ---------- @@ -89,21 +82,22 @@ def _open_writer(path: str) -> sqlite3.Connection: """ connection = sqlite3.connect(path) try: - connection.execute(SCHEMA) + connection.execute(SCHEMA.format(table=self._table)) connection.commit() except sqlite3.DatabaseError as error: connection.close() raise ValueError(f"'{path}' cannot be used as a cache: {error}") from error return connection - @staticmethod - def _open_reader(path: str) -> sqlite3.Connection: + def _open_reader(self, path: str) -> sqlite3.Connection: """ - Opens a cache file read-only and confirms its cache table is usable. + Opens a cache file read-only and confirms it can answer this provider's lookups. - The columns lookups read are checked as well as the table itself, so a - foreign database that happens to hold a differently shaped cache table is - refused here rather than failing mid-run on the first lookup. + A file named as a read-only cache is one the run expects to save calls, + so a file holding nothing for this provider is refused here rather than + quietly costing the calls it was meant to spare. The columns lookups + read are checked as well as the table itself, so a foreign database that + happens to hold a table of that name is refused too. Parameters ---------- @@ -118,23 +112,24 @@ def _open_reader(path: str) -> sqlite3.Connection: Raises ---------- ValueError - If the file is not a SQLite database or holds no usable cache table. + If the file is not a SQLite database, or holds no usable table for + this provider. """ connection = sqlite3.connect(f"{Path(path).resolve().as_uri()}?mode=ro", uri=True) try: - columns = {row[1] for row in connection.execute("PRAGMA table_info(cache)")} + columns = {row[1] for row in connection.execute(f"PRAGMA table_info({self._table})")} except sqlite3.DatabaseError as error: connection.close() raise ValueError(f"'{path}' is not a readable SQLite database: {error}") from error if not columns: connection.close() - raise ValueError(f"'{path}' is not a geocoder cache (it has no 'cache' table)") + raise ValueError(f"'{path}' holds no cached '{self._api}' responses") missing = [column for column in REQUIRED_COLUMNS if column not in columns] if missing: connection.close() - raise ValueError(f"'{path}' is not a geocoder cache (its 'cache' table is missing {', '.join(missing)})") + raise ValueError(f"'{path}' is not a geocoder cache (its '{self._api}' table is missing {', '.join(missing)})") return connection def _connections(self) -> Iterator[sqlite3.Connection]: @@ -143,19 +138,18 @@ def _connections(self) -> Iterator[sqlite3.Connection]: yield self._writer yield from self._readers - def lookup(self, api: str, keys: Iterable[str]) -> Dict[str, Any]: + def lookup(self, keys: Iterable[str]) -> Dict[str, Any]: """ Fetches the stored responses for the given keys, writable cache first. - The first file holding a key wins, and keys with no stored response are - absent from the result. + The first file holding a key under the current version wins; keys with + no such response are absent from the result, whether they were never + stored or were stored under a version this run no longer calls. Parameters ---------- - api : str - The provider name whose entries are searched. keys : Iterable[str] - The normalized query keys to look for. + The query keys to look for. Return ---------- @@ -171,28 +165,31 @@ def lookup(self, api: str, keys: Iterable[str]) -> Dict[str, Any]: for start in range(0, len(outstanding), self.LOOKUP_CHUNK): chunk = outstanding[start : start + self.LOOKUP_CHUNK] placeholders = ",".join("?" * len(chunk)) - rows = connection.execute(f"SELECT query, response FROM cache WHERE api = ? AND query IN ({placeholders})", [api, *chunk]) + rows = connection.execute( + f"SELECT query, response FROM {self._table} WHERE version = ? AND query IN ({placeholders})", + [self._version, *chunk], + ) for key, response in rows: found[key] = json.loads(response) outstanding = [key for key in outstanding if key not in found] return found - def store(self, api: str, key: str, query: str, response: Any) -> None: + def store(self, key: str, response: Any) -> None: """ Records one provider response, committing once a batch has accumulated. + The response is tagged with the version that produced it, replacing any + entry the query already had, so an obsolete response is retired by the + call that supersedes it. + Committing as the run proceeds means a crash costs only the calls made since the last commit. Parameters ---------- - api : str - The provider name the response came from. key : str - The normalized query key to store the response under. - query : str - The query text as the provider composed it, kept for manual review. + The query key to store the response under. response : Any The raw provider response, stored as JSON. """ @@ -201,8 +198,8 @@ def store(self, api: str, key: str, query: str, response: Any) -> None: fetched_at = datetime.now(timezone.utc).isoformat(timespec="seconds") self._writer.execute( - "INSERT OR REPLACE INTO cache (api, query, original_query, response, fetched_at) VALUES (?, ?, ?, ?, ?)", - (api, key, query, json.dumps(response, ensure_ascii=False), fetched_at), + f"INSERT OR REPLACE INTO {self._table} (query, version, response, fetched_at) VALUES (?, ?, ?, ?)", + (key, self._version, json.dumps(response, ensure_ascii=False), fetched_at), ) self._uncommitted += 1 diff --git a/src/census.py b/src/census.py index 966b003..d71cda8 100644 --- a/src/census.py +++ b/src/census.py @@ -7,7 +7,6 @@ import csv import io -import json from typing import Any, Dict, Iterator, List, Tuple import requests @@ -25,7 +24,9 @@ class CensusProvider(Provider): The benchmark and vintage are pinned to the frozen Census2020 dataset so the positional response layout stays stable; the addressbatch endpoint returns - headerless CSV, so columns can only be read by their fixed position. + headerless CSV, so columns can only be read by their fixed position. That + pinned pair is also the version cached responses are tagged with, since + moving to another dataset asks a different question of the service. """ requires_key = False @@ -34,6 +35,7 @@ class CensusProvider(Provider): ENDPOINT = "https://geocoding.geo.census.gov/geocoder/geographies/addressbatch" BENCHMARK = "Public_AR_Census2020" VINTAGE = "Census2020_Census2020" + CACHE_VERSION = f"{BENCHMARK}/{VINTAGE}" BATCH_SIZE = 10000 TIMEOUT = 300 @@ -52,30 +54,6 @@ class CensusProvider(Provider): "block", ] - def cache_key(self, record: SourceRecord) -> str: - """ - Builds the key from the components Census is sent, plus the pinned dataset. - - Country is excluded because the addressbatch CSV has no country column, - while the benchmark and vintage select which dataset answers the query. - - The components are JSON-encoded rather than joined on a separator, so a - row whose fields happen to contain that separator cannot produce the same - key as a different row and be answered with its response. - - Parameters - ---------- - record : SourceRecord - The record whose query is being composed. - - Return - ---------- - str - The query text identifying this record's response. - """ - parts = [record.address, record.city, record.stateprov, record.postalcode, self.BENCHMARK, self.VINTAGE] - return json.dumps(parts, ensure_ascii=False) - def _fetch(self, records: List[SourceRecord]) -> Iterator[Tuple[SourceRecord, Dict[str, Any]]]: """ Posts records in batches, yielding each record with its response row. diff --git a/test/cache_test.py b/test/cache_test.py index 0384dd1..10eadfd 100644 --- a/test/cache_test.py +++ b/test/cache_test.py @@ -12,14 +12,15 @@ import pytest -from src.api import GeocodeResult, Provider, SourceRecord -from src.cache import Cache, missing_files, normalize_query +from src.api import GeocodeResult, Provider, SourceRecord, cache_key +from src.cache import Cache, missing_files class _CountingProvider(Provider): """Counts the records it is asked to fetch so cache hits can be observed.""" name = "counting" + CACHE_VERSION = "v1" def __init__(self, cache=None): super().__init__(cache=cache) @@ -41,15 +42,25 @@ def _record(key, address, city="Town", stateprov="CA"): return SourceRecord(internal_key=key, address=address, city=city, stateprov=stateprov) -def test_normalize_query_folds_case_and_whitespace(): - """Case and runs of whitespace collapse so equivalent queries share one key.""" - assert normalize_query("123 Main St") == normalize_query("123 MAIN ST") - assert normalize_query(" 1 A St\t") == "1 a st" +def _cache(path=None, read_paths=()): + """Opens a cache scoped to the counting provider under test.""" + return Cache(_CountingProvider.name, _CountingProvider.CACHE_VERSION, path, read_paths) -def test_normalize_query_keeps_distinct_addresses_distinct(): +def test_cache_key_folds_case_and_whitespace(): + """Case and runs of whitespace collapse so equivalent rows share one key.""" + assert cache_key(_record(0, "123 Main St")) == cache_key(_record(1, "123 MAIN ST")) + assert cache_key(SourceRecord(internal_key=0, address=" 1 A St\t")) == "1 a st" + + +def test_cache_key_keeps_distinct_addresses_distinct(): """Normalization does not merge addresses that differ in substance.""" - assert normalize_query("1 Main St") != normalize_query("1 Main Ave") + assert cache_key(_record(0, "1 Main St")) != cache_key(_record(1, "1 Main Ave")) + + +def test_cache_key_is_empty_for_a_row_with_nothing_to_query(): + """A row left with no address to send has no query, and so no key.""" + assert cache_key(SourceRecord(internal_key=0, address="01-17-040-06w4")) == "" def test_duplicates_collapse_to_one_call_without_a_cache_file(): @@ -74,12 +85,12 @@ def test_cache_hit_across_runs_skips_the_api(tmp_path): """A second run over an overlapping list only calls the API for the new addresses.""" path = str(tmp_path / "cache.sqlite") - with Cache(path) as cache: + with _cache(path) as cache: first = _CountingProvider(cache) first.geocode([_record(0, "1 Main St"), _record(1, "2 Oak St")]) assert len(first.fetched) == 2 - with Cache(path) as cache: + with _cache(path) as cache: second = _CountingProvider(cache) results = second.geocode([_record(0, "1 Main St"), _record(1, "3 Elm St")]) @@ -92,26 +103,51 @@ def test_cache_is_scoped_per_api(tmp_path): """One provider's stored response is never served to another provider.""" path = str(tmp_path / "cache.sqlite") - with Cache(path) as cache: + with _cache(path) as cache: _CountingProvider(cache).geocode([_record(0, "1 Main St")]) - with Cache(path) as cache: + with Cache("different", "v1", path) as cache: other = _CountingProvider(cache) - other.name = "different" other.geocode([_record(0, "1 Main St")]) assert other.fetched == ["1 Main St, Town, CA"] +def test_cache_is_scoped_per_api_version(tmp_path): + """An entry stored under an earlier version of the API is never served.""" + path = str(tmp_path / "cache.sqlite") + + with _cache(path) as cache: + _CountingProvider(cache).geocode([_record(0, "1 Main St")]) + + with Cache("counting", "v2", path) as cache: + later = _CountingProvider(cache) + later.geocode([_record(0, "1 Main St")]) + + assert later.fetched == ["1 Main St, Town, CA"] + + +def test_refetching_replaces_the_entry_of_an_earlier_version(tmp_path): + """A query re-asked under a new version overwrites its row instead of accumulating one.""" + path = str(tmp_path / "cache.sqlite") + + with _cache(path) as cache: + _CountingProvider(cache).geocode([_record(0, "1 Main St")]) + with Cache("counting", "v2", path) as cache: + _CountingProvider(cache).geocode([_record(0, "1 Main St")]) + + assert sqlite3.connect(path).execute("SELECT version FROM counting").fetchall() == [("v2",)] + + def test_read_only_cache_is_used_but_not_written(tmp_path): """A --cacheRead file answers lookups and is left untouched by the run.""" shared = tmp_path / "shared.sqlite" - with Cache(str(shared)) as cache: + with _cache(str(shared)) as cache: _CountingProvider(cache).geocode([_record(0, "1 Main St")]) before = shared.read_bytes() - with Cache(str(tmp_path / "own.sqlite"), [str(shared)]) as cache: + with _cache(str(tmp_path / "own.sqlite"), [str(shared)]) as cache: provider = _CountingProvider(cache) provider.geocode([_record(0, "1 Main St"), _record(1, "2 Oak St")]) @@ -123,14 +159,14 @@ def test_writable_cache_wins_over_read_only(tmp_path): """The writable cache answers first when both files hold the same key.""" stale = str(tmp_path / "stale.sqlite") fresh = str(tmp_path / "fresh.sqlite") - key = normalize_query("1 Main St, Town, CA") + key = cache_key(_record(0, "1 Main St")) - with Cache(stale) as cache: - cache.store("counting", key, "1 Main St, Town, CA", {"echo": "from stale"}) - with Cache(fresh) as cache: - cache.store("counting", key, "1 Main St, Town, CA", {"echo": "from fresh"}) + with _cache(stale) as cache: + cache.store(key, {"echo": "from stale"}) + with _cache(fresh) as cache: + cache.store(key, {"echo": "from fresh"}) - with Cache(fresh, [stale]) as cache: + with _cache(fresh, [stale]) as cache: result = _CountingProvider(cache).geocode([_record(0, "1 Main St")])[0] assert result.result_address == "from fresh" @@ -139,29 +175,30 @@ def test_writable_cache_wins_over_read_only(tmp_path): def test_store_commits_in_batches_for_crash_safety(tmp_path): """Responses reach disk before the run ends so a crash keeps completed calls.""" path = str(tmp_path / "cache.sqlite") - cache = Cache(path) + cache = _cache(path) for index in range(Cache.COMMIT_INTERVAL): - cache.store("counting", f"key {index}", f"key {index}", {"echo": index}) + cache.store(f"key {index}", {"echo": index}) - committed = sqlite3.connect(path).execute("SELECT COUNT(*) FROM cache").fetchone()[0] + committed = sqlite3.connect(path).execute("SELECT COUNT(*) FROM counting").fetchone()[0] cache.close() assert committed == Cache.COMMIT_INTERVAL -def test_stored_row_keeps_the_original_query(tmp_path): - """The un-normalized query is retained alongside its key for manual review.""" +def test_stored_row_records_the_version_and_the_fetch_time(tmp_path): + """Each entry carries the key it answers, the version behind it, and when it arrived.""" path = str(tmp_path / "cache.sqlite") - with Cache(path) as cache: + with _cache(path) as cache: _CountingProvider(cache).geocode([_record(0, "1 MAIN St")]) - row = sqlite3.connect(path).execute("SELECT query, original_query FROM cache").fetchone() + row = sqlite3.connect(path).execute("SELECT query, version, fetched_at FROM counting").fetchone() assert row[0] == "1 main st, town, ca" - assert row[1] == "1 MAIN St, Town, CA" + assert row[1] == "v1" + assert row[2] -def test_read_only_cache_rejects_a_foreign_database(tmp_path): - """A SQLite file without a cache table is refused instead of failing later.""" +def test_read_only_cache_rejects_a_database_holding_nothing_for_the_provider(tmp_path): + """A file that cannot answer this provider is refused instead of quietly saving nothing.""" path = str(tmp_path / "other.sqlite") connection = sqlite3.connect(path) connection.execute("CREATE TABLE unrelated (id INTEGER)") @@ -169,19 +206,19 @@ def test_read_only_cache_rejects_a_foreign_database(tmp_path): connection.close() with pytest.raises(ValueError): - Cache(None, [path]) + _cache(None, [path]) -def test_read_only_cache_rejects_a_differently_shaped_cache_table(tmp_path): - """A 'cache' table without the columns lookups read is refused up front.""" +def test_read_only_cache_rejects_a_differently_shaped_table(tmp_path): + """A table named for the provider without the columns lookups read is refused up front.""" path = str(tmp_path / "other.sqlite") connection = sqlite3.connect(path) - connection.execute("CREATE TABLE cache (key TEXT, value TEXT)") + connection.execute("CREATE TABLE counting (key TEXT, value TEXT)") connection.commit() connection.close() with pytest.raises(ValueError): - Cache(None, [path]) + _cache(None, [path]) def test_read_only_cache_rejects_a_non_database(tmp_path): @@ -190,7 +227,7 @@ def test_read_only_cache_rejects_a_non_database(tmp_path): path.write_text("not a database") with pytest.raises(ValueError): - Cache(None, [str(path)]) + _cache(None, [str(path)]) def test_writable_cache_rejects_a_non_database(tmp_path): @@ -199,7 +236,7 @@ def test_writable_cache_rejects_a_non_database(tmp_path): path.write_text("not a database") with pytest.raises(ValueError): - Cache(str(path)) + _cache(str(path)) assert path.read_text() == "not a database" diff --git a/test/census_test.py b/test/census_test.py index 917b933..f38ba82 100644 --- a/test/census_test.py +++ b/test/census_test.py @@ -161,24 +161,10 @@ def fake_post(*_args, **_kwargs): census.CensusProvider().geocode(records) -def test_census_cache_key_ignores_country_and_pins_dataset(): - """The key covers the posted components and the dataset, but not the unsent country.""" - provider = census.CensusProvider() - usa = SourceRecord(internal_key=0, address="1 Main St", city="Town", stateprov="CA", postalcode="90210", country="US") - blank = SourceRecord(internal_key=1, address="1 Main St", city="Town", stateprov="CA", postalcode="90210") - - assert provider.cache_key(usa) == provider.cache_key(blank) - assert census.CensusProvider.BENCHMARK in provider.cache_key(usa) - assert census.CensusProvider.VINTAGE in provider.cache_key(usa) - - -def test_census_cache_key_keeps_shifted_components_distinct(): - """Rows whose components differ only in where a field boundary falls get distinct keys.""" - provider = census.CensusProvider() - split = SourceRecord(internal_key=0, address="1 Main St|Town", city="CA") - shifted = SourceRecord(internal_key=1, address="1 Main St", city="Town|CA") - - assert provider.cache_key(split) != provider.cache_key(shifted) +def test_census_cache_version_pins_the_dataset(): + """Cached entries are tagged with the dataset that answered them.""" + assert census.CensusProvider.BENCHMARK in census.CensusProvider.CACHE_VERSION + assert census.CensusProvider.VINTAGE in census.CensusProvider.CACHE_VERSION def test_census_posts_each_distinct_address_once(monkeypatch): From 6e0aee492b79763d14f5ea1996c702dbd7e0f165 Mon Sep 17 00:00:00 2001 From: Zachary Schwartz <112527479+ZachSchwartz@users.noreply.github.com> Date: Mon, 21 Sep 2026 11:58:48 -0500 Subject: [PATCH 7/7] Update cache.py --- src/cache.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/cache.py b/src/cache.py index 07afe4c..377f9b5 100644 --- a/src/cache.py +++ b/src/cache.py @@ -224,7 +224,7 @@ def __enter__(self) -> "Cache": """Returns the cache so it can be used as a context manager.""" return self - def __exit__(self, *_exception) -> None: + def __exit__(self, *_exception: object) -> None: """Closes the cache when the context exits.""" self.close()