junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-08 11:17:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Move updating of is_default to domain rather than database layer

Browse source
This commit is contained in:
Daniel Roth 2026-02-13 09:39:25 +00:00
parent 4b07310d6b
commit f34a6269f7
3 changed files with 92 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
backend/app/db/functions/recommendations_functions.py
View file

@ -632,12 +632,12 @@ def get_scenario(scenario_id: int) -> Optional[ScenarioModel]:
return session_any.exec(stmt).scalar_one_or_none()
def set_plan_default(plan_id: int, is_default: bool) -> bool:
def update_plan(plan_model: PlanModel, scenario_model: ScenarioModel) -> bool:
with db_read_session() as session:
stmt = (
update(PlanModel)
.where(PlanModel.id == plan_id)
.values(is_default=is_default)
.where(PlanModel.id == plan_model.id)
.values(**plan_model.model_dump(exclude={"id"}, exclude_unset=True))
)
result = session.exec(stmt)
session.commit()

78
backend/app/domain/classes/plan.py
View file

@ -2,8 +2,10 @@ from __future__ import annotations
from dataclasses import replace
from typing import Optional
from sqlalchemy import Tuple
from backend.app.db.models.portfolio import PortfolioGoal
from backend.app.db.models.recommendations import PlanModel
from backend.app.db.models.recommendations import PlanModel, ScenarioModel
from backend.app.domain.classes.scenario import Scenario
from backend.app.domain.records.plan_record import PlanRecord
from backend.app.utils import sap_to_epc
@ -56,8 +58,82 @@ class Plan:
case _:
raise NotImplementedError
def to_sqlalchemy(self) -> Tuple[PlanModel, ScenarioModel]:
scenario_record = self.scenario.record
scenario_model = ScenarioModel(
id=self.scenario.id,
name=scenario_record.name,
created_at=scenario_record.created_at,
housing_type=scenario_record.housing_type,
goal=scenario_record.goal,
goal_value=scenario_record.goal_value,
trigger_file_path=scenario_record.trigger_file_path,
multi_plan=scenario_record.multi_plan,
is_default=scenario_record.is_default,
budget=scenario_record.budget,
already_installed_file_path=scenario_record.already_installed_file_path,
patches_file_path=scenario_record.patches_file_path,
non_invasive_recommendations_file_path=scenario_record.non_invasive_recommendations_file_path,
exclusions=scenario_record.exclusions,
cost=scenario_record.cost,
contingency=scenario_record.contingency,
funding=scenario_record.funding,
total_work_hours=scenario_record.total_work_hours,
energy_savings=scenario_record.energy_savings,
co2_equivalent_savings=scenario_record.co2_equivalent_savings,
energy_cost_savings=scenario_record.energy_cost_savings,
epc_breakdown_pre_retrofit=scenario_record.epc_breakdown_pre_retrofit,
epc_breakdown_post_retrofit=scenario_record.epc_breakdown_post_retrofit,
number_of_properties=scenario_record.number_of_properties,
n_units_to_retrofit=scenario_record.n_units_to_retrofit,
co2_per_unit_pre_retrofit=scenario_record.co2_per_unit_pre_retrofit,
co2_per_unit_post_retrofit=scenario_record.co2_per_unit_post_retrofit,
energy_bill_per_unit_pre_retrofit=scenario_record.energy_bill_per_unit_pre_retrofit,
energy_bill_per_unit_post_retrofit=scenario_record.energy_bill_per_unit_post_retrofit,
energy_consumption_per_unit_pre_retrofit=scenario_record.energy_consumption_per_unit_pre_retrofit,
energy_consumption_per_unit_post_retrofit=scenario_record.energy_consumption_per_unit_post_retrofit,
valuation_improvement_per_unit=scenario_record.valuation_improvement_per_unit,
cost_per_unit=scenario_record.cost_per_unit,
cost_per_co2_saved=scenario_record.cost_per_co2_saved,
cost_per_sap_point=scenario_record.cost_per_sap_point,
valuation_return_on_investment=scenario_record.valuation_return_on_investment,
property_valuation_increase=scenario_record.property_valuation_increase,
labour_days=scenario_record.labour_days,
)
record = self.record
plan_model = PlanModel(
id=self.id,
property_id=record.property_id,
portfolio_id=record.portfolio_id,
scenario_id=self.scenario.id,
created_at=record.created_at,
is_default=record.is_default,
valuation_increase_lower_bound=record.valuation_increase_lower_bound,
valuation_increase_upper_bound=record.valuation_increase_upper_bound,
valuation_increase_average=record.valuation_increase_average,
plan_type=record.plan_type,
post_sap_points=record.post_sap_points,
post_epc_rating=record.post_epc_rating,
post_co2_emissions=record.post_co2_emissions,
co2_savings=record.co2_savings,
post_energy_bill=record.post_energy_bill,
energy_bill_savings=record.energy_bill_savings,
post_energy_consumption=record.post_energy_consumption,
energy_consumption_savings=record.energy_consumption_savings,
valuation_post_retrofit=record.valuation_post_retrofit,
valuation_increase=record.valuation_increase,
cost_of_works=record.cost_of_works,
contingency_cost=record.contingency_cost,
)
return Tuple(plan_model, scenario_model) # TODO: create a type for this
def set_default(self, value: bool) -> None:
self.record = replace(self.record, is_default=value)
self.scenario.record = replace(self.scenario.record, is_default=value)
def _is_compliant_epc(self) -> bool:
goal_value: str = self.scenario.record.goal_value

16
backend/categorisation/processor.py
View file

@ -1,11 +1,15 @@
from collections import defaultdict
from typing import List
from typing import List, cast
from sqlalchemy import Tuple
from backend.app.db.functions.recommendations_functions import (
get_plans_by_portfolio_id,
get_scenario,
set_plan_default,
update_plan,
)
from backend.app.db.models.recommendations import PlanModel, ScenarioModel
from backend.app.domain.classes.plan import Plan
from backend.categorisation.categorisation_logic import CategorisationLogic
from utils.logger import setup_logger
@ -58,7 +62,11 @@ def _update_default_flags(plans: List[Plan], cheapest_plan: Plan) -> None:
if plan.id is None:
raise ValueError("Cannot update Plan with missing ID")
set_plan_default(
plan.id,
plan.id == cheapest_plan.id,
plan.set_default(plan.id == cheapest_plan.id)
plan_model, scenario_model = cast(
tuple[PlanModel, ScenarioModel],
plan.to_sqlalchemy(),
)
update_plan(plan_model, scenario_model)