Take a Postcode value object at the historic EPC repository boundary 🟩

The port accepts a normalised Postcode and rejects malformed/empty ones via
Postcode.is_valid() (PostcodeNotFound) — dropping the per-lookup postcodes.io
HTTP call and the cross-module use of the private _sanitise_postcode. Mapper
helper promoted to public.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jun-te Kim 2026-06-29 15:24:18 +00:00
parent 23dfb3f899
commit 0536f70162
7 changed files with 69 additions and 135 deletions

View file

@ -4,7 +4,6 @@ from typing import Optional
import pandas as pd
from backend.address2UPRN.scoring import rank_address_similarity
from backend.utils.addressMatch import AddressMatch
from datatypes.epc.domain.historic_epc import HistoricEpc
from utils.pandas_utils import pandas_cell_to_str
@ -13,7 +12,7 @@ DEFAULT_S3_ROOT = "s3://retrofit-data-dev/historical_epc"
_EXTRA_COLS = {"lexiscore", "lexirank"}
def _map_historic_epc_pandas_row_to_domain(row: pd.Series) -> HistoricEpc:
def map_historic_epc_pandas_row_to_domain(row: pd.Series) -> HistoricEpc:
kwargs = {
col.lower(): pandas_cell_to_str(val)
for col, val in row.items()
@ -92,15 +91,6 @@ def rank_historic_epc(
]
def _sanitise_postcode(postcode: str) -> str:
cleaned = (postcode or "").upper().replace(" ", "")
if not cleaned:
raise ValueError("postcode must contain non-whitespace characters")
if not AddressMatch.is_valid_postcode(cleaned):
raise ValueError(f"postcode {cleaned!r} is not a valid UK postcode")
return cleaned
def match_addresses_for_postcode(
user_address: str,
postcode: str,

View file

@ -6,11 +6,9 @@ import pandas as pd
import pytest
from botocore.exceptions import ClientError
from datatypes.epc.domain import historic_epc_matching as matcher_mod
from datatypes.epc.domain.historic_epc_matching import (
HistoricEpcMatches,
ScoredHistoricEpc,
_sanitise_postcode,
match_addresses_for_postcode,
)
from repositories.historic_epc import historic_epc_s3_repository as repo_mod
@ -126,14 +124,6 @@ def _build_df(rows: list[dict]) -> pd.DataFrame:
return pd.DataFrame(rows, columns=_FULL_COLUMN_FIELDS)
@pytest.fixture
def patch_postcode_valid():
with patch.object(
matcher_mod.AddressMatch, "is_valid_postcode", return_value=True
) as m:
yield m
@pytest.fixture
def patch_read():
# match_addresses_for_postcode now reads via the historic-EPC repository, so
@ -142,38 +132,12 @@ def patch_read():
yield m
# ---------- _sanitise_postcode ----------
class TestSanitisePostcode:
def test_uppercases_and_strips_spaces(self, patch_postcode_valid):
assert _sanitise_postcode("ab33 8al") == "AB338AL"
def test_empty_raises(self, patch_postcode_valid):
with pytest.raises(ValueError, match="non-whitespace"):
_sanitise_postcode("")
def test_whitespace_only_raises(self, patch_postcode_valid):
with pytest.raises(ValueError, match="non-whitespace"):
_sanitise_postcode(" ")
def test_invalid_postcode_raises(self):
with patch.object(
matcher_mod.AddressMatch, "is_valid_postcode", return_value=False
):
with pytest.raises(ValueError, match="not a valid UK postcode"):
_sanitise_postcode("NONSENSE")
# ---------- match_addresses_for_postcode ----------
class TestMatchAddressesForPostcode:
def test_preserves_row_count_including_zero_score_rows(
self, patch_read, patch_postcode_valid
):
def test_preserves_row_count_including_zero_score_rows(self, patch_read):
# Disjoint number sets => hard zero. Still kept in matches.
patch_read.return_value = _build_df(
[
@ -185,9 +149,7 @@ class TestMatchAddressesForPostcode:
assert isinstance(result, HistoricEpcMatches)
assert len(result.matches) == 2
def test_top_has_lexirank_one_and_lexiscore_monotone(
self, patch_read, patch_postcode_valid
):
def test_top_has_lexirank_one_and_lexiscore_monotone(self, patch_read):
patch_read.return_value = _build_df(
[
_row("48 GORDON ROAD", "200"), # near miss
@ -195,20 +157,20 @@ class TestMatchAddressesForPostcode:
]
)
result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
assert result.top().lexirank == 1
top = result.top()
assert top is not None
assert top.lexirank == 1
scores = [m.lexiscore for m in result.matches]
assert scores == sorted(scores, reverse=True)
def test_s3_key_built_from_default_root(self, patch_read, patch_postcode_valid):
def test_s3_key_built_from_default_root(self, patch_read):
patch_read.return_value = _build_df([_row("47 GORDON ROAD", "100")])
match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
patch_read.assert_called_once_with(
"retrofit-data-dev", "historical_epc/AB338AL/data.csv.gz"
)
def test_s3_key_respects_custom_root_with_trailing_slash(
self, patch_read, patch_postcode_valid
):
def test_s3_key_respects_custom_root_with_trailing_slash(self, patch_read):
patch_read.return_value = _build_df([_row("47 GORDON ROAD", "100")])
match_addresses_for_postcode(
"47 Gordon Road",
@ -219,9 +181,7 @@ class TestMatchAddressesForPostcode:
"my-bucket", "some/prefix/AB338AL/data.csv.gz"
)
def test_missing_postcode_object_yields_empty_matches(
self, patch_read, patch_postcode_valid
):
def test_missing_postcode_object_yields_empty_matches(self, patch_read):
# A valid postcode with no stored shard is a normal miss, not an error:
# empty matches, not FileNotFoundError.
patch_read.side_effect = ClientError(
@ -231,14 +191,14 @@ class TestMatchAddressesForPostcode:
assert result.matches == []
assert result.unambiguous_uprn() is None
def test_other_client_error_propagates(self, patch_read, patch_postcode_valid):
def test_other_client_error_propagates(self, patch_read):
patch_read.side_effect = ClientError(
{"Error": {"Code": "AccessDenied", "Message": "nope"}}, "GetObject"
)
with pytest.raises(ClientError):
match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
def test_empty_user_address_raises(self, patch_postcode_valid):
def test_empty_user_address_raises(self):
with pytest.raises(ValueError, match="user_address"):
match_addresses_for_postcode("", "AB33 8AL")
@ -248,7 +208,7 @@ class TestMatchAddressesForPostcode:
class TestUnambiguousUprn:
def test_exact_match_returns_uprn(self, patch_read, patch_postcode_valid):
def test_exact_match_returns_uprn(self, patch_read):
patch_read.return_value = _build_df(
[
_row("47 GORDON ROAD", "100"),
@ -258,7 +218,7 @@ class TestUnambiguousUprn:
result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
assert result.unambiguous_uprn() == "100"
def test_ambiguous_tie_returns_none(self, patch_read, patch_postcode_valid):
def test_ambiguous_tie_returns_none(self, patch_read):
# Two duplicate addresses with different UPRNs share rank-1.
patch_read.return_value = _build_df(
[
@ -269,9 +229,7 @@ class TestUnambiguousUprn:
result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
assert result.unambiguous_uprn() is None
def test_all_zero_score_returns_none_even_when_uprn_unique(
self, patch_read, patch_postcode_valid
):
def test_all_zero_score_returns_none_even_when_uprn_unique(self, patch_read):
# User address has building number 47; no row has 47 -> all hard-zero.
patch_read.return_value = _build_df(
[
@ -283,9 +241,7 @@ class TestUnambiguousUprn:
assert all(m.lexiscore == 0.0 for m in result.matches)
assert result.unambiguous_uprn() is None
def test_nan_uprn_becomes_empty_string_not_nan(
self, patch_read, patch_postcode_valid
):
def test_nan_uprn_becomes_empty_string_not_nan(self, patch_read):
# Use a real NaN in the UPRN cell.
patch_read.return_value = _build_df(
[
@ -310,7 +266,7 @@ class TestUnambiguousUprn:
class TestTopHelpers:
def test_top_n_returns_first_k(self, patch_read, patch_postcode_valid):
def test_top_n_returns_first_k(self, patch_read):
patch_read.return_value = _build_df(
[
_row("47 GORDON ROAD", "100"),

View file

@ -3,15 +3,15 @@ from __future__ import annotations
from abc import ABC, abstractmethod
from datatypes.epc.domain.historic_epc import HistoricEpc
from domain.postcode import Postcode
class PostcodeNotFound(Exception):
"""The postcode is empty or not a valid UK postcode, so it cannot key a
historic-EPC lookup.
"""The postcode is empty, so it cannot key a historic-EPC lookup.
Distinct from a *valid* postcode that simply has no stored data that case
returns an empty list, because a miss is the normal, expected outcome of a
best-effort historic lookup.
Distinct from a non-empty postcode that simply has no stored data that
case returns an empty list, because a miss is the normal, expected outcome
of a best-effort historic lookup.
"""
@ -20,9 +20,11 @@ class HistoricEpcRepository(ABC):
sharded by postcode in S3 (``historical_epc/{POSTCODE}/data.csv.gz``).
A Repo, not a Fetcher (ADR-0011): it reads stored data with no live EPC API
call. A valid postcode with no stored object returns ``[]``; an unusable
postcode raises :class:`PostcodeNotFound`.
call. Takes a normalised :class:`Postcode` value object (so the boundary is
strictly typed and never re-sanitises a raw string). A non-empty postcode
with no stored object returns ``[]``; an empty one raises
:class:`PostcodeNotFound`.
"""
@abstractmethod
def get_for_postcode(self, postcode: str) -> list[HistoricEpc]: ...
def get_for_postcode(self, postcode: Postcode) -> list[HistoricEpc]: ...

View file

@ -6,10 +6,8 @@ from datatypes.epc.domain.historic_epc_matching import (
HistoricEpcMatches,
rank_historic_epc,
)
from repositories.historic_epc.historic_epc_repository import (
HistoricEpcRepository,
PostcodeNotFound,
)
from domain.postcode import Postcode
from repositories.historic_epc.historic_epc_repository import HistoricEpcRepository
class HistoricEpcResolver:
@ -25,11 +23,12 @@ class HistoricEpcResolver:
"""All of the postcode's historic EPCs scored against ``user_address``."""
if not user_address:
raise ValueError("user_address must be non-empty")
records = self._repo.get_for_postcode(postcode)
pc = Postcode(postcode)
records = self._repo.get_for_postcode(pc)
matches = rank_historic_epc(records, user_address)
return HistoricEpcMatches(
user_address=user_address,
postcode=postcode.upper().replace(" ", ""),
postcode=str(pc),
matches=matches,
)

View file

@ -6,10 +6,10 @@ from typing import Optional
import pandas as pd
from botocore.exceptions import ClientError
from domain.postcode import Postcode
from datatypes.epc.domain.historic_epc import HistoricEpc
from datatypes.epc.domain.historic_epc_matching import (
_map_historic_epc_pandas_row_to_domain,
_sanitise_postcode,
map_historic_epc_pandas_row_to_domain,
)
from repositories.historic_epc.historic_epc_repository import (
HistoricEpcRepository,
@ -38,17 +38,15 @@ class HistoricEpcS3Repository(HistoricEpcRepository):
self._read_csv_gz: CsvGzReader = read_csv_gz or read_csv_gz_from_s3
self._s3_root = s3_root
def get_for_postcode(self, postcode: str) -> list[HistoricEpc]:
try:
pc = _sanitise_postcode(postcode)
except ValueError as e:
raise PostcodeNotFound(str(e)) from e
def get_for_postcode(self, postcode: Postcode) -> list[HistoricEpc]:
if not postcode.is_valid():
raise PostcodeNotFound(f"{postcode.value!r} is not a valid UK postcode")
bucket, root_prefix = parse_s3_uri(self._s3_root)
key = f"{root_prefix.rstrip('/')}/{pc}/data.csv.gz"
key = f"{root_prefix.rstrip('/')}/{postcode}/data.csv.gz"
try:
df = self._read_csv_gz(bucket, key)
except ClientError as e:
if e.response.get("Error", {}).get("Code") in ("NoSuchKey", "404"):
return []
raise
return [_map_historic_epc_pandas_row_to_domain(row) for _, row in df.iterrows()]
return [map_historic_epc_pandas_row_to_domain(row) for _, row in df.iterrows()]

View file

@ -11,6 +11,7 @@ import dataclasses
from datatypes.epc.domain.historic_epc import HistoricEpc
from datatypes.epc.domain.historic_epc_matching import HistoricEpcMatches
from domain.postcode import Postcode
from repositories.historic_epc.historic_epc_repository import HistoricEpcRepository
from repositories.historic_epc.historic_epc_resolver import HistoricEpcResolver
@ -26,9 +27,8 @@ class _FakeRepo(HistoricEpcRepository):
def __init__(self, by_postcode: dict[str, list[HistoricEpc]]) -> None:
self._by_postcode = by_postcode
def get_for_postcode(self, postcode: str) -> list[HistoricEpc]:
key = postcode.upper().replace(" ", "")
return self._by_postcode.get(key, [])
def get_for_postcode(self, postcode: Postcode) -> list[HistoricEpc]:
return self._by_postcode.get(str(postcode), [])
def test_match_composes_repo_and_matcher_into_scored_matches():
@ -50,7 +50,9 @@ def test_match_composes_repo_and_matcher_into_scored_matches():
assert isinstance(result, HistoricEpcMatches)
assert result.postcode == "AB338AL"
assert len(result.matches) == 2
assert result.top().record.address == "47 GORDON ROAD"
top = result.top()
assert top is not None
assert top.record.address == "47 GORDON ROAD"
def test_resolve_uprn_returns_unambiguous_match():

View file

@ -1,23 +1,22 @@
"""HistoricEpcS3Repository reads per-postcode shards of the old-EPC backup.
A reference-data lookup, not a Fetcher (ADR-0011): no live EPC API call. The
adapter reads ``historical_epc/{POSTCODE}/data.csv.gz`` via an injected reader,
so the tests exercise mapping + key construction + absence against a fake reader
with no network.
adapter takes a normalised ``Postcode`` and reads
``historical_epc/{POSTCODE}/data.csv.gz`` via an injected reader, so the tests
exercise mapping + key construction + absence against a fake reader with no
network.
"""
from __future__ import annotations
import dataclasses
from contextlib import contextmanager
from unittest.mock import patch
import pandas as pd
import pytest
from botocore.exceptions import ClientError
from backend.utils.addressMatch import AddressMatch
from datatypes.epc.domain.historic_epc import HistoricEpc
from domain.postcode import Postcode
from repositories.historic_epc.historic_epc_repository import PostcodeNotFound
from repositories.historic_epc.historic_epc_s3_repository import (
HistoricEpcS3Repository,
@ -28,31 +27,24 @@ from repositories.historic_epc.historic_epc_s3_repository import (
_COLS = [f.name.upper() for f in dataclasses.fields(HistoricEpc)]
def _row(address: str, uprn: object) -> dict:
row = {col: "" for col in _COLS}
def _row(address: str, uprn: object) -> dict[str, object]:
row: dict[str, object] = {col: "" for col in _COLS}
row["ADDRESS"] = address
row["UPRN"] = uprn
return row
def _df(rows: list[dict]) -> pd.DataFrame:
def _df(rows: list[dict[str, object]]) -> pd.DataFrame:
return pd.DataFrame(rows, columns=_COLS)
@contextmanager
def _valid_postcode():
with patch.object(AddressMatch, "is_valid_postcode", return_value=True):
yield
def test_get_for_postcode_maps_reader_rows_to_historic_epc_records():
# Arrange
df = _df([_row("47 GORDON ROAD", "100")])
repo = HistoricEpcS3Repository(read_csv_gz=lambda bucket, key: df)
# Act
with _valid_postcode():
records = repo.get_for_postcode("AB33 8AL")
records = repo.get_for_postcode(Postcode("AB33 8AL"))
# Assert
assert len(records) == 1
@ -61,8 +53,8 @@ def test_get_for_postcode_maps_reader_rows_to_historic_epc_records():
assert records[0].uprn == "100"
def test_valid_postcode_with_no_stored_object_returns_empty_list():
# Arrange — a valid postcode whose shard does not exist in S3.
def test_non_empty_postcode_with_no_stored_object_returns_empty_list():
# Arrange — a postcode whose shard does not exist in S3.
def missing(bucket: str, key: str) -> pd.DataFrame:
raise ClientError(
{"Error": {"Code": "NoSuchKey", "Message": "missing"}}, "GetObject"
@ -71,14 +63,13 @@ def test_valid_postcode_with_no_stored_object_returns_empty_list():
repo = HistoricEpcS3Repository(read_csv_gz=missing)
# Act
with _valid_postcode():
records = repo.get_for_postcode("AB33 8AL")
records = repo.get_for_postcode(Postcode("AB33 8AL"))
# Assert — a miss is the normal, expected outcome, not an exception.
assert records == []
def test_builds_s3_key_from_sanitised_postcode_and_default_root():
def test_builds_s3_key_from_postcode_and_default_root():
# Arrange
calls: list[tuple[str, str]] = []
@ -88,9 +79,8 @@ def test_builds_s3_key_from_sanitised_postcode_and_default_root():
repo = HistoricEpcS3Repository(read_csv_gz=capture)
# Act
with _valid_postcode():
repo.get_for_postcode("ab33 8al")
# Act — the Postcode value object has already normalised the casing/spacing.
repo.get_for_postcode(Postcode("ab33 8al"))
# Assert
assert calls == [("retrofit-data-dev", "historical_epc/AB338AL/data.csv.gz")]
@ -106,28 +96,26 @@ def test_non_missing_read_error_propagates():
repo = HistoricEpcS3Repository(read_csv_gz=denied)
# Act / Assert
with _valid_postcode():
with pytest.raises(ClientError):
repo.get_for_postcode("AB33 8AL")
with pytest.raises(ClientError):
repo.get_for_postcode(Postcode("AB33 8AL"))
def test_empty_postcode_raises_postcode_not_found():
# Arrange
# Arrange — Postcode normalises whitespace away, leaving an empty key.
repo = HistoricEpcS3Repository(read_csv_gz=lambda bucket, key: _df([]))
# Act / Assert — an unusable key, distinct from a valid-but-absent postcode.
# Act / Assert — an unusable key, distinct from a non-empty absent postcode.
with pytest.raises(PostcodeNotFound):
repo.get_for_postcode(" ")
repo.get_for_postcode(Postcode(" "))
def test_invalid_postcode_raises_postcode_not_found():
# Arrange
def test_malformed_postcode_raises_postcode_not_found():
# Arrange — a non-empty but malformed postcode can't key a real shard.
repo = HistoricEpcS3Repository(read_csv_gz=lambda bucket, key: _df([]))
# Act / Assert
with patch.object(AddressMatch, "is_valid_postcode", return_value=False):
with pytest.raises(PostcodeNotFound):
repo.get_for_postcode("NONSENSE")
with pytest.raises(PostcodeNotFound):
repo.get_for_postcode(Postcode("NONSENSE"))
def test_uprn_trailing_dot_zero_is_stripped():
@ -137,8 +125,7 @@ def test_uprn_trailing_dot_zero_is_stripped():
repo = HistoricEpcS3Repository(read_csv_gz=lambda bucket, key: df)
# Act
with _valid_postcode():
records = repo.get_for_postcode("AB33 8AL")
records = repo.get_for_postcode(Postcode("AB33 8AL"))
# Assert
assert records[0].uprn == "151020766"