feat(property): hydrate planning restrictions from the spatial cache

Slice 3c.5. `PropertyPostgresRepository` takes an injected `SpatialRepository`
and hydrates `Property.planning_restrictions` by UPRN (bulk in `get_many`,
single in `get`). A UPRN with no cached row — or a property with no UPRN —
defaults to unrestricted, matching legacy `empty_spatial_df` (ADR-0020). This
closes the loop: Ingestion caches the protections, Modelling reads them off the
Property to gate solid-wall EWI/IWI (ADR-0019).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

repositories/postgres_unit_of_work.py

@@ -40,10 +40,13 @@ class PostgresUnitOfWork(UnitOfWork):
     def __enter__(self) -> "PostgresUnitOfWork":
         self._session = self._session_factory()
         epc_repo = EpcPostgresRepository(self._session)
-        self.property = PropertyPostgresRepository(self._session, epc_repo)
+        spatial_repo = SpatialPostgresRepository(self._session)
+        self.property = PropertyPostgresRepository(
+            self._session, epc_repo, spatial_repo
+        )
         self.epc = epc_repo
         self.solar = SolarPostgresRepository(self._session)
-        self.spatial = SpatialPostgresRepository(self._session)
+        self.spatial = spatial_repo
         self.property_baseline = PropertyBaselinePostgresRepository(self._session)
         self.scenario = ScenarioPostgresRepository(self._session)
         self.product = ProductPostgresRepository(self._session)

repositories/property/property_postgres_repository.py

@@ -1,24 +1,35 @@
 from __future__ import annotations
 
+from typing import Optional
+
 from sqlmodel import Session, col, select
 
+from domain.geospatial.planning_restrictions import PlanningRestrictions
 from domain.property.properties import Properties
 from domain.property.property import Property, PropertyIdentity
 from infrastructure.postgres.property_table import PropertyRow
 from repositories.epc.epc_repository import EpcRepository
 from repositories.property.property_repository import PropertyRepository
+from repositories.spatial.spatial_repository import SpatialRepository
 
 
 class PropertyPostgresRepository(PropertyRepository):
     """Hydrates the Property aggregate from the FE-owned ``property`` row plus the
-    EPC slice (via an injected `EpcRepository`). Reads only from repos — no
-    external IO — so a hydrated Property is a pure function of repository state
-    (ADR-0003).
+    EPC slice (via an injected `EpcRepository`) and the planning protections
+    (via an injected `SpatialRepository`, keyed by UPRN — ADR-0020). Reads only
+    from repos — no external IO — so a hydrated Property is a pure function of
+    repository state (ADR-0003).
     """
 
-    def __init__(self, session: Session, epc_repo: EpcRepository) -> None:
+    def __init__(
+        self,
+        session: Session,
+        epc_repo: EpcRepository,
+        spatial_repo: SpatialRepository,
+    ) -> None:
         self._session = session
         self._epc_repo = epc_repo
+        self._spatial_repo = spatial_repo
 
     def get(self, property_id: int) -> Property:
         row = self._session.get(PropertyRow, property_id)
@@ -31,9 +42,13 @@ class PropertyPostgresRepository(PropertyRepository):
             uprn=row.uprn,
             landlord_property_id=row.landlord_property_id,
         )
+        restrictions: dict[int, PlanningRestrictions] = self._restrictions_for(
+            [row.uprn] if row.uprn is not None else []
+        )
         return Property(
             identity=identity,
             epc=self._epc_repo.get_for_property(property_id),
+            planning_restrictions=_restrictions_of(row.uprn, restrictions),
         )
 
     def get_many(self, property_ids: list[int]) -> Properties:
@@ -44,6 +59,9 @@ class PropertyPostgresRepository(PropertyRepository):
         ).all()
         row_by_id = {row.id: row for row in rows}
         epcs = self._epc_repo.get_for_properties(property_ids)
+        restrictions: dict[int, PlanningRestrictions] = self._restrictions_for(
+            [row.uprn for row in rows if row.uprn is not None]
+        )
         items: list[Property] = []
         for property_id in property_ids:
             row = row_by_id.get(property_id)
@@ -59,6 +77,24 @@ class PropertyPostgresRepository(PropertyRepository):
                         landlord_property_id=row.landlord_property_id,
                     ),
                     epc=epcs.get(property_id),
+                    planning_restrictions=_restrictions_of(row.uprn, restrictions),
                 )
             )
         return Properties(items)
+
+    def _restrictions_for(
+        self, uprns: list[int]
+    ) -> dict[int, PlanningRestrictions]:
+        if not uprns:
+            return {}
+        return self._spatial_repo.get_for_uprns(uprns)
+
+
+def _restrictions_of(
+    uprn: Optional[int], by_uprn: dict[int, PlanningRestrictions]
+) -> PlanningRestrictions:
+    """The cached protections for a UPRN, defaulting to unrestricted when the
+    UPRN is absent or uncached (per legacy `empty_spatial_df`; ADR-0020)."""
+    if uprn is None:
+        return PlanningRestrictions()
+    return by_uprn.get(uprn, PlanningRestrictions())

tests/repositories/property/test_property_repository.py

@@ -10,11 +10,14 @@ from sqlalchemy import Engine
 from sqlmodel import Session
 
 from datatypes.epc.domain.mapper import EpcPropertyDataMapper
+from domain.geospatial.planning_restrictions import PlanningRestrictions
+from domain.geospatial.spatial_reference import SpatialReference
 from infrastructure.postgres.property_table import PropertyRow
 from repositories.epc.epc_postgres_repository import EpcPostgresRepository
 from repositories.property.property_postgres_repository import (
     PropertyPostgresRepository,
 )
+from repositories.spatial.spatial_postgres_repository import SpatialPostgresRepository
 
 _JSON_SAMPLES = Path(__file__).resolve().parents[3] / "backend/epc_api/json_samples"
 
@@ -38,7 +41,9 @@ def test_get_hydrates_identity_and_epc_slice(db_engine: Engine) -> None:
 
     # Act
     with Session(db_engine) as session:
-        repo = PropertyPostgresRepository(session, EpcPostgresRepository(session))
+        repo = PropertyPostgresRepository(
+            session, EpcPostgresRepository(session), SpatialPostgresRepository(session)
+        )
         prop = repo.get(property_id)
 
     # Assert
@@ -47,3 +52,62 @@ def test_get_hydrates_identity_and_epc_slice(db_engine: Engine) -> None:
     assert prop.epc == epc
     assert prop.source_path == "epc_with_overlay"
     assert prop.effective_epc == epc
+
+
+def test_get_many_hydrates_planning_restrictions_from_the_spatial_cache(
+    db_engine: Engine,
+) -> None:
+    # Arrange — a property whose UPRN has a cached listed-building flag.
+    with Session(db_engine) as session:
+        row = PropertyRow(
+            portfolio_id=7, postcode="A0 0AA", address="1 Some Street", uprn=12345
+        )
+        session.add(row)
+        session.commit()
+        property_id = row.id
+        assert property_id is not None
+        SpatialPostgresRepository(session).save(
+            uprn=12345,
+            reference=SpatialReference(
+                coordinates=None,
+                restrictions=PlanningRestrictions(is_listed=True),
+            ),
+        )
+        session.commit()
+
+    # Act
+    with Session(db_engine) as session:
+        repo = PropertyPostgresRepository(
+            session, EpcPostgresRepository(session), SpatialPostgresRepository(session)
+        )
+        properties = repo.get_many([property_id])
+
+    # Assert — the protections are hydrated onto the Property (ADR-0020).
+    assert properties.items[0].planning_restrictions == PlanningRestrictions(
+        is_listed=True
+    )
+
+
+def test_get_many_defaults_to_unrestricted_when_uprn_has_no_spatial_row(
+    db_engine: Engine,
+) -> None:
+    # Arrange — a property whose UPRN is not in the spatial cache.
+    with Session(db_engine) as session:
+        row = PropertyRow(
+            portfolio_id=7, postcode="A0 0AA", address="1 Some Street", uprn=999
+        )
+        session.add(row)
+        session.commit()
+        property_id = row.id
+        assert property_id is not None
+
+    # Act
+    with Session(db_engine) as session:
+        repo = PropertyPostgresRepository(
+            session, EpcPostgresRepository(session), SpatialPostgresRepository(session)
+        )
+        properties = repo.get_many([property_id])
+
+    # Assert — an uncovered UPRN means unrestricted, not blocked (per legacy
+    # `empty_spatial_df`; ADR-0020).
+    assert properties.items[0].planning_restrictions == PlanningRestrictions()