mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-27 23:35:01 +00:00
104 lines
3.9 KiB
Python
104 lines
3.9 KiB
Python
"""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)."""
|
|
if property_ids:
|
|
return sorted(set(property_ids))
|
|
resolved = resolve_filtered_property_ids(
|
|
self._session, portfolio_id, filters
|
|
)
|
|
return sorted({filtered.property_id for filtered in resolved})
|
|
|
|
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
|