Resolve export property ids with explicit ids overriding filters 🟥

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

backend/app/exports/export_tasks.py

@@ -0,0 +1,99 @@
+"""The Scenario Export route's view of the app-owned task's sub_task (ADR-0065).
+
+Parameterised SQL rather than the SQLModel mirrors: importing the
+``infrastructure.postgres`` mirror of ``sub_task`` into the FastAPI process
+double-registers the table and crashes at import (the same constraint the
+Modelling Run distributor and the bulk-download route work around).
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from typing import Any, cast
+from uuid import UUID
+
+from sqlalchemy import text
+from sqlmodel import Session
+
+from backend.app.modelling.property_filters import (
+    PropertyGroupFilters,
+    resolve_filtered_property_ids,
+)
+
+
+class ScenarioExportTasks:
+    def __init__(self, session: Session) -> None:
+        self._session = session
+
+    def read_selection_config(self, task_id: UUID) -> dict[str, Any]:
+        """The task's selection recipe, read from the FE-owned ``task.inputs``
+        JSON (ADR-0065) — ``{portfolio_id, scenario_ids, filters?, property_ids?,
+        recipient_email?}``. Empty dict when the task has no inputs."""
+        row = (
+            self._session.connection()
+            .execute(
+                text("SELECT inputs FROM tasks WHERE id = :task_id"),
+                {"task_id": task_id},
+            )
+            .first()
+        )
+        if row is None or row[0] is None:
+            return {}
+        parsed: Any = json.loads(row[0])
+        return cast(dict[str, Any], parsed) if isinstance(parsed, dict) else {}
+
+    def already_distributed(self, task_id: UUID) -> bool:
+        """Whether the task already has a sub_task — the 409 guard against a
+        double-submit re-triggering the same export."""
+        row = (
+            self._session.connection()
+            .execute(
+                text("SELECT 1 FROM sub_task WHERE task_id = :task_id LIMIT 1"),
+                {"task_id": task_id},
+            )
+            .first()
+        )
+        return row is not None
+
+    def resolve_property_ids(
+        self,
+        *,
+        portfolio_id: int,
+        filters: PropertyGroupFilters,
+        property_ids: list[int],
+    ) -> list[int]:
+        """Resolve the selection to the distinct internal ``property.id`` set the
+        export is built from (ADR-0065). An explicit ``property_ids`` list is an
+        **override**, not a co-filter: when present it is used as given;
+        otherwise the ``filters`` are resolved against the portfolio through the
+        shared Modelling Run resolver (ADR-0056)."""
+        raise NotImplementedError
+
+    def create_export_subtask(
+        self, task_id: UUID, subtask_id: UUID, recipe: dict[str, Any]
+    ) -> bool:
+        """Pre-create the single ``waiting`` sub_task under the app-owned task,
+        pinning the recipe (resolved ``property_ids``, ``scenario_ids``,
+        ``recipient_email``, ``export_name``) onto its inputs (ADR-0055/0065).
+
+        The insert is conditional on the task having no sub_task yet, so the DB
+        arbitrates a double-submit race. Returns whether this call created it."""
+        now = datetime.now(timezone.utc)
+        result = self._session.connection().execute(
+            text(
+                "INSERT INTO sub_task (id, task_id, status, inputs, updated_at)"
+                " SELECT :id, :task_id, 'waiting', :inputs, :updated_at"
+                " WHERE NOT EXISTS ("
+                "   SELECT 1 FROM sub_task WHERE task_id = :task_id"
+                " )"
+            ),
+            {
+                "id": subtask_id,
+                "task_id": task_id,
+                "inputs": json.dumps(recipe),
+                "updated_at": now,
+            },
+        )
+        self._session.commit()
+        return result.rowcount == 1

tests/backend/app/exports/test_export_tasks.py

@@ -0,0 +1,49 @@
+"""ScenarioExportTasks property-id resolution (ADR-0065): an explicit
+property_ids list overrides the filters; otherwise the shared Modelling Run
+filter resolver (ADR-0056) resolves the portfolio's filtered set."""
+
+from __future__ import annotations
+
+from sqlalchemy import Engine
+from sqlmodel import Session
+
+from backend.app.exports.export_tasks import ScenarioExportTasks
+from backend.app.modelling.property_filters import PropertyGroupFilters
+from infrastructure.postgres.property_table import PropertyRow  # registers `property`
+
+
+def test_explicit_property_ids_override_the_filters(db_engine: Engine) -> None:
+    # arrange — filters that match nothing, but an explicit hand-picked list.
+    with Session(db_engine) as session:
+        # act
+        result = ScenarioExportTasks(session).resolve_property_ids(
+            portfolio_id=100,
+            filters=PropertyGroupFilters(postcodes=["ZZ1 1ZZ"]),
+            property_ids=[3, 1, 2, 1],
+        )
+
+    # assert — the hand-picked ids win (deduplicated, sorted); filters ignored.
+    assert result == [1, 2, 3]
+
+
+def test_resolves_from_filters_when_no_explicit_ids(db_engine: Engine) -> None:
+    # arrange — two properties in the portfolio, one matching the postcode filter.
+    with Session(db_engine) as session:
+        session.add(
+            PropertyRow(id=1, portfolio_id=100, postcode="AB1 2CD", landlord_property_id="LP1")
+        )
+        session.add(
+            PropertyRow(id=2, portfolio_id=100, postcode="XY9 9ZZ", landlord_property_id="LP2")
+        )
+        session.commit()
+
+    # act
+    with Session(db_engine) as session:
+        result = ScenarioExportTasks(session).resolve_property_ids(
+            portfolio_id=100,
+            filters=PropertyGroupFilters(postcodes=["AB1 2CD"]),
+            property_ids=[],
+        )
+
+    # assert — only the property matching the filter is resolved.
+    assert result == [1]