Model/infrastructure/postgres/scenario_table.py
Khalim Conn-Kowlessar 62a968119c feat(modelling): domain Scenario + ScenarioPostgresRepository (#1157)
Slice 1 of the #1157 build. The FE creates a Scenario and passes only
its id to the pipeline; the Modelling stage reads it back here.

- domain/modelling/scenario.py: thin `Scenario(id, goal, goal_value,
  budget, is_default)` — the slice the stage uses today (goal/budget for
  the Optimiser later; is_default drives plan.is_default). No phases
  (ADR-0005); legacy file-path/aggregate columns not modelled.
- infrastructure/postgres/scenario_table.py: `ScenarioRow` SQLModel
  mirror of the live `scenario` table (ADR-0017), declaring only the
  read columns; goal mapped as its string value.
- ScenarioPostgresRepository.get_many(scenario_ids) -> list[Scenario]:
  bulk read, input-order-preserving, raises on a missing id.

The method shape lives on the concrete repo for now; it is promoted to
an @abstractmethod on the port when the real orchestrator is wired and
the bare-stub instantiations retire (keeps the stubbed Modelling wiring
composing meanwhile). 2 tests, pyright strict clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-03 11:19:52 +00:00

41 lines
1.4 KiB
Python

from __future__ import annotations
from typing import ClassVar, Optional
from sqlmodel import Field, SQLModel
from domain.modelling.scenario import Scenario
class ScenarioRow(SQLModel, table=True):
"""SQLModel mirror of the live ``scenario`` table (ADR-0017).
Declares only the columns the Modelling stage reads — the legacy
file-path columns (`trigger_file_path`, `exclusions`, …) and the
portfolio-level aggregates are left to the legacy SQLAlchemy model
(`backend/app/db/models/recommendations.py::ScenarioModel`), which still
owns the live reads. The physical table is the shared contract; this
mirror is read-only from the rebuild's side.
`goal` is a Postgres enum in production; mapped here as its string value
(the Modelling stage does not yet branch on it — #1160).
"""
__tablename__: ClassVar[str] = "scenario" # pyright: ignore[reportIncompatibleVariableOverride]
id: Optional[int] = Field(default=None, primary_key=True)
goal: str
goal_value: str
budget: Optional[float] = Field(default=None)
is_default: bool = Field(default=False)
def to_domain(self) -> Scenario:
if self.id is None:
raise ValueError("scenario row has no id")
return Scenario(
id=self.id,
goal=self.goal,
goal_value=self.goal_value,
budget=self.budget,
is_default=self.is_default,
)