diff --git a/backend/app/main.py b/backend/app/main.py index 0dd9e489b..4b4aaaa70 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -11,6 +11,7 @@ from backend.app.whlg import router as whlg_router from backend.app.plan import router as plan_router from backend.app.tasks import router as tasks_router from backend.app.bulk_uploads import router as bulk_uploads_router +from backend.app.modelling import router as modelling_router from backend.app.dependencies import validate_api_key from backend.app.config import get_settings @@ -63,6 +64,7 @@ app.include_router(portfolio_router.router, prefix="/v1") app.include_router(plan_router.router, prefix="/v1") app.include_router(whlg_router.router, prefix="/v1") app.include_router(bulk_uploads_router.router, prefix="/v1") +app.include_router(modelling_router.router, prefix="/v1") if get_settings().ENVIRONMENT == "local": from backend.app.local import router as local_router diff --git a/backend/app/modelling/property_filters.py b/backend/app/modelling/property_filters.py index ce398cb25..0c0934fb0 100644 --- a/backend/app/modelling/property_filters.py +++ b/backend/app/modelling/property_filters.py @@ -4,18 +4,21 @@ The resolution rule is a shared contract with the app's preview (ADR-0056): the "N properties will be modelled" count the user approves is computed by the Next.js app from this same rule, so any change here must land in both codebases and amend the ADR. + +Reads go through parameterised raw SQL rather than the SQLModel table mirrors: +the FastAPI app registers the legacy ``backend.app.db.models`` mirrors of +``property``/``sub_task``, and importing the ``infrastructure.postgres`` +mirrors of the same tables into one process double-registers them in the +shared SQLModel metadata and crashes the app at import. Until the DDD +cut-over unifies the stacks, this module stays model-free. """ -from collections.abc import Callable from dataclasses import dataclass -from typing import Optional +from typing import Any, Optional from pydantic import BaseModel, ConfigDict -from sqlmodel import Session, col, select - -from infrastructure.postgres.epc_property_table import EpcPropertyModel -from infrastructure.postgres.property_override_table import PropertyOverrideRow -from infrastructure.postgres.property_table import PropertyRow +from sqlalchemy import text +from sqlmodel import Session class PropertyGroupFilters(BaseModel): @@ -48,13 +51,13 @@ _UNKNOWN = "Unknown" class _ComponentResolution: """How one filterable component (property_type / built_form) resolves at each ADR-0056 precedence level: which override row names it, how its RdSAP - numeric codes map (text labels pass through as-is), and where it lives on - the EPC and the legacy property row.""" + numeric codes map (text labels pass through as-is), and which columns hold + it on the EPC and the legacy property row.""" override_component: str codes: dict[str, str] - epc_value: Callable[[EpcPropertyModel], Optional[str]] - legacy_value: Callable[[PropertyRow], Optional[str]] + epc_columns: tuple[str, ...] + legacy_column: str _PROPERTY_TYPE = _ComponentResolution( @@ -67,8 +70,8 @@ _PROPERTY_TYPE = _ComponentResolution( "4": "Park home", }, # A cert without the property_type column falls back to its dwelling_type. - epc_value=lambda epc: epc.property_type or epc.dwelling_type, - legacy_value=lambda row: row.property_type, + epc_columns=("property_type", "dwelling_type"), + legacy_column="property_type", ) _BUILT_FORM = _ComponentResolution( @@ -81,8 +84,8 @@ _BUILT_FORM = _ComponentResolution( "5": "Enclosed End-Terrace", "6": "Enclosed Mid-Terrace", }, - epc_value=lambda epc: epc.built_form, - legacy_value=lambda row: row.built_form, + epc_columns=("built_form",), + legacy_column="built_form", ) @@ -93,17 +96,29 @@ def resolve_filtered_property_ids( portfolio's rows with ``marked_for_deletion = false``; property_type / built_form resolve override → lodged EPC → predicted EPC → legacy columns → "Unknown". Filters AND-combine; an absent key is unconstrained.""" - statement = ( - select(PropertyRow) - .where(PropertyRow.portfolio_id == portfolio_id) - # IS NOT TRUE rather than == False: the FE schema defaults the flag, - # but a NULL must never silently exclude a property. - .where(col(PropertyRow.marked_for_deletion).isnot(True)) - .order_by(col(PropertyRow.id)) + # IS NOT TRUE rather than = false: the FE schema defaults the flag, but a + # NULL must never silently exclude a property. + base_sql = ( + "SELECT id, postcode, property_type, built_form FROM property" + " WHERE portfolio_id = :portfolio_id" + " AND marked_for_deletion IS NOT TRUE" ) + params: dict[str, Any] = {"portfolio_id": portfolio_id} if filters.postcodes is not None: - statement = statement.where(col(PropertyRow.postcode).in_(filters.postcodes)) - candidates = [row for row in session.exec(statement).all() if row.id is not None] + base_sql += " AND postcode = ANY(:postcodes)" + params["postcodes"] = filters.postcodes + base_sql += " ORDER BY id" + candidates = [ + _Candidate( + property_id=int(row[0]), + postcode=row[1], + legacy_values={ + _PROPERTY_TYPE.legacy_column: row[2], + _BUILT_FORM.legacy_column: row[3], + }, + ) + for row in session.connection().execute(text(base_sql), params) + ] for wanted_values, component in ( (filters.property_types, _PROPERTY_TYPE), @@ -114,60 +129,65 @@ def resolve_filtered_property_ids( wanted = set(wanted_values) resolved = _resolved_component_values(session, candidates, component) candidates = [ - row - for row in candidates - if row.id is not None and resolved.get(row.id) in wanted + c for c in candidates if resolved.get(c.property_id) in wanted ] return [ - FilteredProperty(property_id=row.id, postcode=row.postcode) - for row in candidates - if row.id is not None + FilteredProperty(property_id=c.property_id, postcode=c.postcode) + for c in candidates ] +@dataclass(frozen=True) +class _Candidate: + property_id: int + postcode: Optional[str] + legacy_values: dict[str, Optional[str]] + + def _resolved_component_values( - session: Session, candidates: list[PropertyRow], component: _ComponentResolution + session: Session, candidates: list[_Candidate], component: _ComponentResolution ) -> dict[int, str]: """Each candidate's effective value for *component* per the ADR-0056 precedence: override (building_part 0) → lodged EPC → predicted EPC → legacy column → "Unknown".""" - property_ids = [row.id for row in candidates if row.id is not None] + property_ids = [c.property_id for c in candidates] - overrides = session.exec( - select(PropertyOverrideRow) - .where(col(PropertyOverrideRow.property_id).in_(property_ids)) - .where(PropertyOverrideRow.building_part == 0) - .where(PropertyOverrideRow.override_component == component.override_component) - ).all() - by_override = {o.property_id: o.override_value for o in overrides} + override_rows = session.connection().execute( + text( + "SELECT property_id, override_value FROM property_overrides" + " WHERE property_id = ANY(:ids)" + " AND building_part = 0" + " AND override_component = :component" + ), + {"ids": property_ids, "component": component.override_component}, + ) + by_override = {int(row[0]): str(row[1]) for row in override_rows} - epcs = session.exec( - select(EpcPropertyModel) - .where(col(EpcPropertyModel.property_id).in_(property_ids)) - .where(col(EpcPropertyModel.source).in_(("lodged", "predicted"))) - ).all() + epc_columns = ", ".join(component.epc_columns) + epc_rows = session.connection().execute( + text( + f"SELECT property_id, source, {epc_columns} FROM epc_property" + " WHERE property_id = ANY(:ids)" + " AND source IN ('lodged', 'predicted')" + ), + {"ids": property_ids}, + ) by_epc_source: dict[str, dict[int, str]] = {"lodged": {}, "predicted": {}} - for epc in epcs: - value = component.epc_value(epc) - if epc.property_id is not None and value: - by_epc_source[epc.source][epc.property_id] = component.codes.get( - value, value + for row in epc_rows: + value = next((v for v in row[2:] if v), None) + if value is not None: + by_epc_source[str(row[1])][int(row[0])] = component.codes.get( + str(value), str(value) ) - by_legacy_column = { - row.id: value - for row in candidates - if row.id is not None and (value := component.legacy_value(row)) is not None - } - return { - pid: ( - by_override.get(pid) - or by_epc_source["lodged"].get(pid) - or by_epc_source["predicted"].get(pid) - or by_legacy_column.get(pid) + c.property_id: ( + by_override.get(c.property_id) + or by_epc_source["lodged"].get(c.property_id) + or by_epc_source["predicted"].get(c.property_id) + or c.legacy_values.get(component.legacy_column) or _UNKNOWN ) - for pid in property_ids + for c in candidates } diff --git a/backend/app/modelling/router.py b/backend/app/modelling/router.py index 09a9f5bb3..dcb8fbebf 100644 --- a/backend/app/modelling/router.py +++ b/backend/app/modelling/router.py @@ -9,11 +9,14 @@ fan-out — progress and terminal state roll up from the workers. import json from collections.abc import Callable, Iterator +from datetime import datetime, timezone from typing import Any, cast +from uuid import uuid4 import boto3 from fastapi import APIRouter, Depends, HTTPException -from sqlmodel import Session, col, select +from sqlalchemy import text +from sqlmodel import Session from backend.app.config import get_settings from backend.app.db.connection import db_engine @@ -24,9 +27,12 @@ from backend.app.modelling.property_filters import ( resolve_filtered_property_ids, ) from backend.app.modelling.schemas import TriggerRunRequest -from domain.tasks.subtasks import SubTask -from infrastructure.postgres.modelling import ScenarioModel -from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository + +# DB access is parameterised raw SQL, not the SQLModel mirrors: the FastAPI +# app registers the legacy backend.app.db.models mirrors of these tables, and +# importing the infrastructure.postgres mirrors of the same tables into one +# process double-registers them and crashes the app at import (see +# property_filters — same containment, until the DDD cut-over). # Sends pre-serialised message bodies to the modelling_e2e queue. A seam so # tests record bodies instead of calling AWS. @@ -71,8 +77,11 @@ async def trigger_run( session: Session = Depends(get_session), send_messages: MessageSender = Depends(get_message_sender), ) -> dict[str, str]: - subtask_repo = SubTaskPostgresRepository(session) - if subtask_repo.list_by_task(body.task_id): + already_distributed = session.connection().execute( + text("SELECT 1 FROM sub_task WHERE task_id = :task_id LIMIT 1"), + {"task_id": body.task_id}, + ).first() + if already_distributed is not None: raise HTTPException( status_code=409, detail=( @@ -81,10 +90,11 @@ async def trigger_run( ), ) - scenario_rows = session.exec( - select(ScenarioModel).where(col(ScenarioModel.id).in_(body.scenario_ids)) - ).all() - portfolio_by_scenario = {row.id: row.portfolio_id for row in scenario_rows} + scenario_rows = session.connection().execute( + text("SELECT id, portfolio_id FROM scenario WHERE id = ANY(:ids)"), + {"ids": body.scenario_ids}, + ) + portfolio_by_scenario = {int(row[0]): row[1] for row in scenario_rows} invalid = [ scenario_id for scenario_id in body.scenario_ids @@ -117,30 +127,44 @@ async def trigger_run( # Pre-create one sub_task per (scenario, batch) message under the # app-owned task, each holding its exact message payload — the fixed # progress denominator and the batch's re-run recipe (ADR-0055). - subtasks: list[SubTask] = [] - payloads: list[dict[str, Any]] = [] + messages: list[dict[str, Any]] = [] for scenario_id in body.scenario_ids: for batch in batches: - payload: dict[str, Any] = { - "task_id": str(body.task_id), - "property_ids": [p.property_id for p in batch], - "portfolio_id": body.portfolio_id, - "scenario_id": scenario_id, - # ADR-0055 pinned flags: live EPC fetch, live prediction, - # solar fetched only where no stored row exists. - "refetch_epc": True, - "repredict_epc": True, - "refetch_solar": True, - "dry_run": False, - } - subtasks.append(SubTask.create(task_id=body.task_id, inputs=payload)) - payloads.append(payload) - subtask_repo.create_many(subtasks) + messages.append( + { + "task_id": str(body.task_id), + "subtask_id": str(uuid4()), + "property_ids": [p.property_id for p in batch], + "portfolio_id": body.portfolio_id, + "scenario_id": scenario_id, + # ADR-0055 pinned flags: live EPC fetch, live prediction, + # solar fetched only where no stored row exists. + "refetch_epc": True, + "repredict_epc": True, + "refetch_solar": True, + "dry_run": False, + } + ) - send_messages( + now = datetime.now(timezone.utc) + session.connection().execute( + text( + "INSERT INTO sub_task (id, task_id, status, inputs, updated_at)" + " VALUES (:id, :task_id, 'waiting', :inputs, :updated_at)" + ), [ - json.dumps({**payload, "subtask_id": str(subtask.id)}) - for payload, subtask in zip(payloads, subtasks) - ] + { + "id": message["subtask_id"], + "task_id": body.task_id, + "inputs": json.dumps( + {k: v for k, v in message.items() if k != "subtask_id"} + ), + "updated_at": now, + } + for message in messages + ], ) + session.commit() + + send_messages([json.dumps(message) for message in messages]) return {"message": "Modelling Run distributed"}