mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
65 lines
2.4 KiB
Python
65 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
from sqlmodel import Session, col, update
|
|
|
|
from domain.modelling.plan import Plan
|
|
from infrastructure.postgres.modelling import PlanModel, RecommendationModel
|
|
from repositories.plan.plan_repository import PlanRepository
|
|
|
|
|
|
class PlanPostgresRepository(PlanRepository):
|
|
"""Maps a Plan and its Plan Measures onto the live ``plan`` /
|
|
``recommendation`` tables (ADR-0017). Does not commit — the Unit of Work
|
|
owns the transaction (ADR-0012).
|
|
|
|
A re-run INSERTs a fresh Plan rather than deleting the prior one (the cascade
|
|
delete was slow); when the new Plan is the default it demotes any prior
|
|
default Plan for the same (property_id, scenario_id) to ``is_default=False``,
|
|
so readers can select the current Plan via ``is_default=True``."""
|
|
|
|
def __init__(self, session: Session) -> None:
|
|
self._session = session
|
|
|
|
def save(
|
|
self,
|
|
plan: Plan,
|
|
*,
|
|
property_id: int,
|
|
scenario_id: int,
|
|
portfolio_id: int,
|
|
is_default: bool,
|
|
) -> int:
|
|
# Soft-replace (ADR-0012): keep prior Plans as history rather than DELETEing
|
|
# them — the cascade delete of recommendation rows was the slow part. When
|
|
# this Plan is the default, demote every prior Plan for the same
|
|
# (property_id, scenario_id) to is_default=False, so exactly one Plan for
|
|
# the pair stays default (the one just inserted).
|
|
if is_default:
|
|
self._session.exec( # type: ignore[call-overload]
|
|
update(PlanModel)
|
|
.where(
|
|
col(PlanModel.property_id) == property_id,
|
|
col(PlanModel.scenario_id) == scenario_id,
|
|
)
|
|
.values(is_default=False)
|
|
)
|
|
|
|
plan_row = PlanModel.from_domain(
|
|
plan,
|
|
property_id=property_id,
|
|
scenario_id=scenario_id,
|
|
portfolio_id=portfolio_id,
|
|
is_default=is_default,
|
|
)
|
|
self._session.add(plan_row)
|
|
self._session.flush()
|
|
if plan_row.id is None:
|
|
raise ValueError("plan row did not receive an id")
|
|
|
|
for measure in plan.measures:
|
|
self._session.add(
|
|
RecommendationModel.from_domain(
|
|
measure, property_id=property_id, plan_id=plan_row.id
|
|
)
|
|
)
|
|
return plan_row.id
|