Map historic EPC S3 shard to domain records 🟥

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

repositories/historic_epc/historic_epc_repository.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from datatypes.epc.domain.historic_epc import HistoricEpc
+
+
+class PostcodeNotFound(Exception):
+    """The postcode is empty or not a valid UK postcode, 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.
+    """
+
+
+class HistoricEpcRepository(ABC):
+    """Reads the 'old EPC' backup — one flat ``HistoricEpc`` row per certificate,
+    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`.
+    """
+
+    @abstractmethod
+    def get_for_postcode(self, postcode: str) -> list[HistoricEpc]: ...

repositories/historic_epc/historic_epc_s3_repository.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+
+import pandas as pd
+
+from datatypes.epc.domain.historic_epc import HistoricEpc
+from repositories.historic_epc.historic_epc_repository import HistoricEpcRepository
+from utils.s3 import read_csv_gz_from_s3
+
+DEFAULT_S3_ROOT = "s3://retrofit-data-dev/historical_epc"
+
+# (bucket, key) -> DataFrame. Injected so the dataset is sourced from S3 in
+# production or a fake in tests — the Repo holds no S3/HTTP code of its own
+# (mirrors GeospatialS3Repository).
+CsvGzReader = Callable[[str, str], pd.DataFrame]
+
+
+class HistoricEpcS3Repository(HistoricEpcRepository):
+    """Reads per-postcode ``data.csv.gz`` shards of the historic EPC backup."""
+
+    def __init__(
+        self,
+        read_csv_gz: CsvGzReader = read_csv_gz_from_s3,
+        s3_root: str = DEFAULT_S3_ROOT,
+    ) -> None:
+        self._read_csv_gz = read_csv_gz
+        self._s3_root = s3_root
+
+    def get_for_postcode(self, postcode: str) -> list[HistoricEpc]:
+        raise NotImplementedError

tests/repositories/historic_epc/test_historic_epc_s3_repository.py

@@ -0,0 +1,58 @@
+"""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.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+from contextlib import contextmanager
+from unittest.mock import patch
+
+import pandas as pd
+
+from backend.utils.addressMatch import AddressMatch
+from datatypes.epc.domain.historic_epc import HistoricEpc
+from repositories.historic_epc.historic_epc_s3_repository import (
+    HistoricEpcS3Repository,
+)
+
+# HistoricEpc requires every CSV column; derive the (upper-cased) column list
+# straight from the dataclass so it can never drift from the domain type.
+_COLS = [f.name.upper() for f in dataclasses.fields(HistoricEpc)]
+
+
+def _row(address: str, uprn: object) -> dict:
+    row = {col: "" for col in _COLS}
+    row["ADDRESS"] = address
+    row["UPRN"] = uprn
+    return row
+
+
+def _df(rows: list[dict]) -> 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")
+
+    # Assert
+    assert len(records) == 1
+    assert isinstance(records[0], HistoricEpc)
+    assert records[0].address == "47 GORDON ROAD"
+    assert records[0].uprn == "100"