Read a Property's resolved landlord overrides as a faithful value-space snapshot 🟩

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

repositories/property/property_overrides_postgres_reader.py

@@ -0,0 +1,45 @@
+"""Postgres adapter for the ``property_overrides`` read side.
+
+Read-only and uow-independent: ``property_overrides`` is committed reference
+data the ``bulk_upload_finaliser`` Lambda writes at Finalise, long before First
+Run executes — there is no transactional coupling to the ingestion run, so this
+opens its own short read session per call via the injected session factory
+(mirroring the composition root's ``lambda: Session(engine)``).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+
+from sqlmodel import Session, col, select
+
+from infrastructure.postgres.property_override_table import PropertyOverrideRow
+from repositories.property.property_overrides_reader import (
+    PropertyOverridesReader,
+    ResolvedPropertyOverride,
+    ResolvedPropertyOverrides,
+)
+
+
+class PropertyOverridesPostgresReader(PropertyOverridesReader):
+    def __init__(self, session_factory: Callable[[], Session]) -> None:
+        self._session_factory = session_factory
+
+    def overrides_for(self, property_id: int) -> ResolvedPropertyOverrides:
+        with self._session_factory() as session:
+            rows = session.exec(
+                select(PropertyOverrideRow).where(
+                    col(PropertyOverrideRow.property_id) == property_id
+                )
+            ).all()
+
+        return ResolvedPropertyOverrides(
+            rows=tuple(
+                ResolvedPropertyOverride(
+                    override_component=row.override_component,
+                    building_part=row.building_part,
+                    override_value=row.override_value,
+                )
+                for row in rows
+            )
+        )

repositories/property/property_overrides_reader.py

@@ -0,0 +1,52 @@
+"""Read port for the per-Property ``property_overrides`` fact layer (ADR-0006).
+
+The write side (``PropertyOverrideRepository.upsert_all``) materialises the fact
+layer at Finalise; this is the read side. It is deliberately *faithful* — it
+returns the resolved enum-value snapshots exactly as stored ("House",
+"Detached", "Solid brick, …"), every ``(override_component, building_part)`` for
+the Property, making no judgement about what is resolvable. Consumers translate:
+EPC Prediction maps property_type/built_form into the gov-EPC code space and
+gates on it (see ``domain/epc/override_code_mapping.py``); the later
+``epc_with_overlay`` slice will read wall/roof here too.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass(frozen=True)
+class ResolvedPropertyOverride:
+    """One ``property_overrides`` row, in enum-value space (as stored)."""
+
+    override_component: str
+    building_part: int
+    override_value: str
+
+
+@dataclass(frozen=True)
+class ResolvedPropertyOverrides:
+    """Every resolved override for one Property — a faithful value-space snapshot."""
+
+    rows: tuple[ResolvedPropertyOverride, ...]
+
+    def value(self, override_component: str, building_part: int) -> Optional[str]:
+        """The resolved value for one ``(component, building_part)``, or None when
+        the Property has no such override row."""
+        for row in self.rows:
+            if (
+                row.override_component == override_component
+                and row.building_part == building_part
+            ):
+                return row.override_value
+        return None
+
+
+class PropertyOverridesReader(ABC):
+    @abstractmethod
+    def overrides_for(self, property_id: int) -> ResolvedPropertyOverrides:
+        """Every resolved Landlord Override for the Property, as stored. Empty when
+        the Property has no overrides."""
+        ...

tests/repositories/property/test_property_overrides_postgres_reader.py

@@ -0,0 +1,117 @@
+"""Integration tests for the ``property_overrides`` read adapter.
+
+The reader is *faithful*: it returns the resolved enum-value snapshots exactly
+as the finaliser wrote them, every ``(override_component, building_part)`` for
+the Property, with no translation or gating. Verified against a real Postgres
+(the ``db_engine`` fixture) because the value is in reading what was actually
+persisted.
+"""
+
+from __future__ import annotations
+
+from sqlalchemy import Engine
+from sqlmodel import Session
+
+from infrastructure.postgres.property_override_table import PropertyOverrideRow
+from repositories.property.property_overrides_postgres_reader import (
+    PropertyOverridesPostgresReader,
+)
+
+
+def _seed(
+    session: Session,
+    *,
+    override_component: str,
+    override_value: str,
+    property_id: int = 1,
+    portfolio_id: int = 1,
+    building_part: int = 0,
+) -> None:
+    row = PropertyOverrideRow(
+        property_id=property_id,
+        portfolio_id=portfolio_id,
+        building_part=building_part,
+        override_component=override_component,
+        override_value=override_value,
+        original_spreadsheet_description="detached house",
+    )
+    session.add(row)
+
+
+def test_reads_a_resolved_override_in_value_space(db_engine: Engine) -> None:
+    # Arrange — the finaliser wrote one property_type override for the Property.
+    with Session(db_engine) as session:
+        _seed(
+            session,
+            property_id=42,
+            override_component="property_type",
+            override_value="House",
+        )
+        session.commit()
+
+    reader = PropertyOverridesPostgresReader(lambda: Session(db_engine))
+
+    # Act
+    resolved = reader.overrides_for(42)
+
+    # Assert — the stored enum value, untranslated.
+    assert resolved.value("property_type", 0) == "House"
+
+
+def test_returns_every_component_and_building_part_for_the_property(
+    db_engine: Engine,
+) -> None:
+    # Arrange — three overrides across two building parts for the target Property,
+    # plus an override for a *different* Property that must not leak in.
+    with Session(db_engine) as session:
+        _seed(
+            session,
+            property_id=7,
+            building_part=0,
+            override_component="property_type",
+            override_value="House",
+        )
+        _seed(
+            session,
+            property_id=7,
+            building_part=0,
+            override_component="built_form_type",
+            override_value="Detached",
+        )
+        _seed(
+            session,
+            property_id=7,
+            building_part=1,
+            override_component="wall_type",
+            override_value="Solid brick, with internal insulation",
+        )
+        _seed(
+            session,
+            property_id=8,
+            override_component="property_type",
+            override_value="Flat",
+        )
+        session.commit()
+
+    reader = PropertyOverridesPostgresReader(lambda: Session(db_engine))
+
+    # Act
+    resolved = reader.overrides_for(7)
+
+    # Assert — all three of the Property's rows, faithfully; none from Property 8.
+    assert len(resolved.rows) == 3
+    assert resolved.value("property_type", 0) == "House"
+    assert resolved.value("built_form_type", 0) == "Detached"
+    assert resolved.value("wall_type", 1) == "Solid brick, with internal insulation"
+
+
+def test_property_without_overrides_reads_empty(db_engine: Engine) -> None:
+    # Arrange — nothing seeded for this Property.
+    reader = PropertyOverridesPostgresReader(lambda: Session(db_engine))
+
+    # Act
+    resolved = reader.overrides_for(999)
+
+    # Assert
+    assert resolved.rows == ()
+    assert resolved.value("property_type", 0) is None