give classes immutable records to protect udpating

This commit is contained in:
Daniel Roth 2026-02-12 12:58:44 +00:00
parent a0515ea3bb
commit 4ddb5592f3
4 changed files with 118 additions and 27 deletions

View file

@ -0,0 +1,46 @@
from __future__ import annotations
from dataclasses import replace
from typing import Optional
from backend.app.db.models.recommendations import PlanModel
from backend.app.domain.classes.scenario import Scenario
from backend.app.domain.records.plan_record import PlanRecord
class Plan:
def __init__(
self, record: PlanRecord, scenario: Scenario, id: Optional[int] = None
):
self.id = id
self._record = record
self.scenario = scenario
@classmethod
def from_sqlalchemy(cls, plan_model: PlanModel, scenario: Scenario) -> Plan:
record = PlanRecord(
property_id=plan_model.property_id,
portfolio_id=plan_model.portfolio_id,
scenario_id=plan_model.scenario_id,
created_at=plan_model.created_at,
is_default=plan_model.is_default,
valuation_increase_lower_bound=plan_model.valuation_increase_lower_bound,
valuation_increase_upper_bound=plan_model.valuation_increase_upper_bound,
valuation_increase_average=plan_model.valuation_increase_average,
plan_type=plan_model.plan_type,
post_sap_points=plan_model.post_sap_points,
post_epc_rating=plan_model.post_epc_rating,
post_co2_emissions=plan_model.post_co2_emissions,
co2_savings=plan_model.co2_savings,
post_energy_bill=plan_model.post_energy_bill,
energy_bill_savings=plan_model.energy_bill_savings,
post_energy_consumption=plan_model.post_energy_consumption,
energy_consumption_savings=plan_model.energy_consumption_savings,
valuation_post_retrofit=plan_model.valuation_post_retrofit,
valuation_increase=plan_model.valuation_increase,
cost_of_works=plan_model.cost_of_works,
contingency_cost=plan_model.contingency_cost,
)
return cls(record=record, scenario=scenario, id=plan_model.id)
def set_default(self, value: bool) -> None:
self._record = replace(self._record, is_default=value)

View file

@ -0,0 +1,58 @@
from __future__ import annotations
from dataclasses import replace
from typing import Optional
from backend.app.db.models.recommendations import ScenarioModel
from backend.app.domain.records.scenario_record import ScenarioRecord
class Scenario:
def __init__(self, record: ScenarioRecord, id: Optional[int] = None):
self.id = id
self._record = record
@classmethod
def from_sqlalchemy(cls, scenario_model: ScenarioModel) -> Scenario:
record = ScenarioRecord(
name=scenario_model.name,
created_at=scenario_model.created_at,
housing_type=scenario_model.housing_type,
goal=scenario_model.goal,
goal_value=scenario_model.goal_value,
trigger_file_path=scenario_model.trigger_file_path,
multi_plan=scenario_model.multi_plan,
is_default=scenario_model.is_default,
budget=scenario_model.budget,
already_installed_file_path=scenario_model.already_installed_file_path,
patches_file_path=scenario_model.patches_file_path,
non_invasive_recommendations_file_path=scenario_model.non_invasive_recommendations_file_path,
exclusions=scenario_model.exclusions,
cost=scenario_model.cost,
contingency=scenario_model.contingency,
funding=scenario_model.funding,
total_work_hours=scenario_model.total_work_hours,
energy_savings=scenario_model.energy_savings,
co2_equivalent_savings=scenario_model.co2_equivalent_savings,
energy_cost_savings=scenario_model.energy_cost_savings,
epc_breakdown_pre_retrofit=scenario_model.epc_breakdown_pre_retrofit,
epc_breakdown_post_retrofit=scenario_model.epc_breakdown_post_retrofit,
number_of_properties=scenario_model.number_of_properties,
n_units_to_retrofit=scenario_model.n_units_to_retrofit,
co2_per_unit_pre_retrofit=scenario_model.co2_per_unit_pre_retrofit,
co2_per_unit_post_retrofit=scenario_model.co2_per_unit_post_retrofit,
energy_bill_per_unit_pre_retrofit=scenario_model.energy_bill_per_unit_pre_retrofit,
energy_bill_per_unit_post_retrofit=scenario_model.energy_bill_per_unit_post_retrofit,
energy_consumption_per_unit_pre_retrofit=scenario_model.energy_consumption_per_unit_pre_retrofit,
energy_consumption_per_unit_post_retrofit=scenario_model.energy_consumption_per_unit_post_retrofit,
valuation_improvement_per_unit=scenario_model.valuation_improvement_per_unit,
cost_per_unit=scenario_model.cost_per_unit,
cost_per_co2_saved=scenario_model.cost_per_co2_saved,
cost_per_sap_point=scenario_model.cost_per_sap_point,
valuation_return_on_investment=scenario_model.valuation_return_on_investment,
property_valuation_increase=scenario_model.property_valuation_increase,
labour_days=scenario_model.labour_days,
)
return cls(record, scenario_model.id)
def set_default(self, value: bool) -> None:
self._record = replace(self._record, is_default=value)

View file

@ -1,16 +1,16 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
from backend.app.db.models.portfolio import Epc
from backend.app.db.models.recommendations import PlanModel, PlanTypeEnum, ScenarioModel
from backend.app.domain.scenario import Scenario
from backend.app.db.models.recommendations import PlanTypeEnum
class Plan:
@dataclass(frozen=True)
class PlanRecord:
property_id: int
portfolio_id: int
scenario: Scenario
scenario_id: Optional[int]
created_at: datetime
is_default: bool
@ -23,15 +23,10 @@ class Plan:
post_co2_emissions: Optional[float] = None
co2_savings: Optional[float] = None
post_energy_bill: Optional[float] = None
energy_bill_savings: Optional[float] = None
post_energy_consumption: Optional[float] = None
energy_consumption_savings: Optional[float] = None
valuation_post_retrofit: Optional[float] = None
valuation_increase: Optional[float] = None
cost_of_works: Optional[float] = None
contingency_cost: Optional[float] = None
@classmethod
def from_sqlalchemy(
cls, plan_model: PlanModel, scenario_model: ScenarioModel
) -> Plan:
raise NotImplementedError

View file

@ -1,28 +1,24 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
from backend.app.db.models.recommendations import ScenarioModel
class Scenario:
@dataclass(frozen=True)
class ScenarioRecord:
name: str
created_at: datetime
housing_type: str
goal: str # TODO: make enum
goal: str
goal_value: str
trigger_file_path: str
multi_plan: bool
is_default: bool # TODO: isn't this Plan-level?
is_default: bool
budget: Optional[float] = None
already_installed_file_path: Optional[str] = None
patches_file_path: Optional[str] = None
non_invasive_recommendations_file_path: Optional[str] = None
exclusions: Optional[str] = None
# Previously portfolio-level fields
# TODO: are these needed scenario-level?
cost: Optional[float] = None
contingency: Optional[float] = None
funding: Optional[float] = None
@ -30,8 +26,8 @@ class Scenario:
energy_savings: Optional[float] = None
co2_equivalent_savings: Optional[float] = None
energy_cost_savings: Optional[float] = None
epc_breakdown_pre_retrofit: Optional[int] = None
epc_breakdown_post_retrofit: Optional[int] = None
epc_breakdown_pre_retrofit: Optional[str] = None
epc_breakdown_post_retrofit: Optional[str] = None
number_of_properties: Optional[int] = None
n_units_to_retrofit: Optional[int] = None
co2_per_unit_pre_retrofit: Optional[str] = None
@ -44,10 +40,6 @@ class Scenario:
cost_per_unit: Optional[str] = None
cost_per_co2_saved: Optional[str] = None
cost_per_sap_point: Optional[str] = None
valuation_return_on_ivestment: Optional[str] = None
valuation_return_on_investment: Optional[str] = None
property_valuation_increase: Optional[float] = None
labour_days: Optional[float] = None
@classmethod
def from_sqlalchemy(cls, scenario_model: ScenarioModel) -> Scenario:
raise NotImplementedError