mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
13b75bcb0a
JsonCachingEpcClient caches at the raw layer (cert payloads verbatim, search rows as JSON) so warm re-runs replay through retry+mapper+parsing in minutes AND the cache doubles as integration-test fixture material — the ADR-0030 corpus pattern (anonymised raw JSON through the mapper), which pickled domain objects could never be. Wired into the pairs harness (--cache-dir). roof_construction_code_sweep pins the RdSAP roof code table the basement-code-6 way: single-building-part certs only, code x description-prefix purity table. Also fixes the tfa telemetry key rename that left the agreement row empty. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
81 lines
3 KiB
Python
81 lines
3 KiB
Python
"""A raw-JSON disk cache around the EPC-API client, for harness scripts.
|
|
|
|
Caches at the RAW layer — certificate payloads exactly as the API returned
|
|
them, search results as plain JSON rows — so that:
|
|
|
|
1. **Iteration is fast**: the pairs harness re-reads the same certs on every
|
|
run (pairs and cohorts don't change), so each re-run was re-paying 10-20k
|
|
sequential HTTP calls; a warm cache replays in minutes.
|
|
2. **The cache is fixture material**: a committed integration-test corpus
|
|
should hold anonymised raw API JSON loaded through
|
|
``EpcPropertyDataMapper.from_api_response`` (the ADR-0030 corpus pattern),
|
|
so replays keep exercising the mapper and survive domain-dataclass changes.
|
|
Pickles of internal dataclasses would do neither.
|
|
|
|
Successful responses only — errors propagate uncached. Script-support code,
|
|
not a repository: production ingestion must keep reading live (a re-lodgement
|
|
must be seen), so this stays out of the DDD layers.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any, Optional, cast
|
|
|
|
from datatypes.epc.search.epc_search_result import EpcSearchResult
|
|
from infrastructure.epc_client.epc_client_service import EpcClientService
|
|
|
|
|
|
def _safe(key: str) -> str:
|
|
return re.sub(r"[^A-Za-z0-9_-]", "_", key)
|
|
|
|
|
|
class JsonCachingEpcClient(EpcClientService):
|
|
"""An ``EpcClientService`` whose raw HTTP layer is disk-cached as JSON.
|
|
|
|
Overrides the two private fetchers, so retry, mapping and search parsing
|
|
all still run on every call — a warm replay exercises the exact production
|
|
code path minus the network.
|
|
"""
|
|
|
|
def __init__(self, auth_token: str, cache_dir: Path) -> None:
|
|
super().__init__(auth_token)
|
|
self._cache_dir = cache_dir
|
|
cache_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def _path(self, key: str) -> Path:
|
|
return self._cache_dir / f"{_safe(key)}.json"
|
|
|
|
def _read(self, key: str) -> Optional[Any]:
|
|
path = self._path(key)
|
|
if not path.exists():
|
|
return None
|
|
return json.loads(path.read_text())
|
|
|
|
def _write(self, key: str, value: Any) -> None:
|
|
self._path(key).write_text(json.dumps(value))
|
|
|
|
def _fetch_certificate(self, cert_num: str) -> dict[str, Any]:
|
|
cached = self._read(f"cert_{cert_num}")
|
|
if cached is not None:
|
|
return cast(dict[str, Any], cached)
|
|
raw = super()._fetch_certificate(cert_num)
|
|
self._write(f"cert_{cert_num}", raw)
|
|
return raw
|
|
|
|
def _search(
|
|
self,
|
|
postcode: Optional[str] = None,
|
|
uprn: Optional[int] = None,
|
|
) -> list[EpcSearchResult]:
|
|
key = f"search_pc_{postcode}" if postcode else f"search_uprn_{uprn}"
|
|
cached = self._read(key)
|
|
if cached is not None:
|
|
rows = cast(list[dict[str, Any]], cached)
|
|
return [EpcSearchResult(**row) for row in rows]
|
|
results = super()._search(postcode=postcode, uprn=uprn)
|
|
self._write(key, [dataclasses.asdict(r) for r in results])
|
|
return results
|