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

Don't re-get scenarios for every plan

Browse source
This commit is contained in:
Daniel Roth 2026-02-17 15:25:51 +00:00
parent 50a8629cba
commit fac418adbe
2 changed files with 20 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -625,6 +625,13 @@ def get_plans_by_portfolio_id(portfolio_id: int) -> List[PlanModel]:
return session_any.exec(stmt).scalars().all()
def get_scenarios_by_portfolio_id(portfolio_id: int) -> List[ScenarioModel]:
stmt = select(ScenarioModel).where(ScenarioModel.portfolio_id == portfolio_id)
with db_read_session() as session:
session_any: Any = session # Typehint as Any to satisfy Pylance...
return session_any.exec(stmt).scalars().all()
def get_scenario(scenario_id: int) -> Optional[ScenarioModel]:
stmt = select(ScenarioModel).where(ScenarioModel.id == scenario_id)
with db_read_session() as session:

21
backend/categorisation/processor.py
View file

@ -4,7 +4,7 @@ from typing import Dict, List
from backend.app.db.functions.recommendations_functions import (
bulk_update_plans,
get_plans_by_portfolio_id,
get_scenario,
get_scenarios_by_portfolio_id,
)
from backend.app.db.models.recommendations import PlanModel, ScenarioModel
from backend.app.domain.classes.plan import Plan
@ -15,7 +15,7 @@ logger = setup_logger()
def process_portfolio(portfolio_id: int) -> None:
print(f"Processing portfolio {portfolio_id}")
logger.info(f"Processing portfolio {portfolio_id}")
plans: List[Plan] = _load_plans_for_portfolio(portfolio_id)
plans_by_property: Dict[int, List[Plan]] = _group_plans_by_property(plans)
@ -29,22 +29,27 @@ def process_portfolio(portfolio_id: int) -> None:
def _load_plans_for_portfolio(portfolio_id: int) -> List[Plan]:
plan_models = get_plans_by_portfolio_id(portfolio_id)
print(f"Got {len(plan_models)} plans from database")
plans: List[Plan] = []
plan_models = get_plans_by_portfolio_id(portfolio_id)
scenarios: List[ScenarioModel] = get_scenarios_by_portfolio_id(portfolio_id)
if not scenarios:
raise Exception(f"No scenarios found for Portfolio {portfolio_id}")
for model in plan_models:
if not model.scenario_id:
scenario_model = next((s for s in scenarios if s.id == model.scenario_id))
if not scenario_model:
logger.info(f"No Scenario associated with Plan of ID {model.id}")
continue
scenario_model = get_scenario(model.scenario_id)
plans.append(
Plan.from_sqlalchemy(model, Scenario.from_sqlalchemy(scenario_model))
)
print("Successfully mapped plan and scenario to domain object")
logger.info("Successfully mapped plan and scenario to domain object")
logger.info(f"Got {len(plans)} plans from database")
return plans