diff --git a/repositories/historic_epc/__init__.py b/repositories/historic_epc/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/repositories/historic_epc/historic_epc_repository.py b/repositories/historic_epc/historic_epc_repository.py new file mode 100644 index 000000000..30439dcc4 --- /dev/null +++ b/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]: ... diff --git a/repositories/historic_epc/historic_epc_s3_repository.py b/repositories/historic_epc/historic_epc_s3_repository.py new file mode 100644 index 000000000..095ae989f --- /dev/null +++ b/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 diff --git a/tests/repositories/historic_epc/__init__.py b/tests/repositories/historic_epc/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/repositories/historic_epc/test_historic_epc_s3_repository.py b/tests/repositories/historic_epc/test_historic_epc_s3_repository.py new file mode 100644 index 000000000..4524d75af --- /dev/null +++ b/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"