mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
112 lines
3.9 KiB
Python
112 lines
3.9 KiB
Python
from sqlalchemy import insert
|
|
from backend.app.db.models.recommendations import Plan, Recommendation, RecommendationMaterials, PlanRecommendations
|
|
|
|
|
|
def create_plan(session, plan):
|
|
"""
|
|
This function will create a record for the plan in the database if it does not exist.
|
|
:param plan: dictionary of data representing a plan to be created
|
|
"""
|
|
|
|
new_plan = Plan(**plan)
|
|
session.add(new_plan)
|
|
session.flush()
|
|
|
|
return new_plan.id
|
|
|
|
|
|
def create_recommendation(session, recommendation):
|
|
"""
|
|
This function will create a record for the recommendation in the database if it does not exist.
|
|
:param session: The database session
|
|
:param recommendation: dictionary of data representing a recommendation to be created
|
|
"""
|
|
|
|
new_recommendation = Recommendation(**recommendation)
|
|
session.add(new_recommendation)
|
|
session.flush()
|
|
|
|
return new_recommendation.id
|
|
|
|
|
|
def create_recommendation_material(session, recommendation_id, material_id, depth):
|
|
"""
|
|
This function will create a record for the recommendation_material in the database if it does not exist.
|
|
:param session: The databse session
|
|
:param recommendation_id: ID of the recommendation
|
|
:param material_id: ID of the material
|
|
:param depth: depth of the material, may be null if a material where depth is not applicable
|
|
"""
|
|
|
|
new_recommendation_material = RecommendationMaterials(
|
|
recommendation_id=recommendation_id,
|
|
material_id=material_id,
|
|
depth=depth
|
|
)
|
|
session.add(new_recommendation_material)
|
|
session.flush()
|
|
|
|
return new_recommendation_material.id
|
|
|
|
|
|
def create_plan_recommendations(session, plan_id, recommendation_ids):
|
|
"""
|
|
This function will create records for the plan_recommendation in the database.
|
|
:param plan_id: ID of the plan
|
|
:param recommendation_ids: list of recommendation IDs
|
|
"""
|
|
|
|
# Prepare a list of dictionaries for bulk insert
|
|
data = [{"plan_id": plan_id, "recommendation_id": rid} for rid in recommendation_ids]
|
|
|
|
# Bulk insert using SQLAlchemy's core API
|
|
session.execute(insert(PlanRecommendations).values(data))
|
|
|
|
|
|
def upload_recommendations(session, recommendations_to_upload, property_id):
|
|
# Prepare data for bulk insert for Recommendation
|
|
recommendations_data = [
|
|
{
|
|
"property_id": property_id,
|
|
"type": rec["type"],
|
|
"description": rec["description"],
|
|
"estimated_cost": rec["cost"],
|
|
"default": rec["default"],
|
|
"starting_u_value": rec.get("starting_u_value"),
|
|
"new_u_value": rec.get("new_u_value"),
|
|
"sap_points": rec["sap_points"]
|
|
}
|
|
for rec in recommendations_to_upload
|
|
]
|
|
|
|
session.bulk_insert_mappings(Recommendation, recommendations_data)
|
|
|
|
# To get the IDs of the newly inserted recommendations, we need to flush the session
|
|
session.flush()
|
|
|
|
# Map the uploaded_recommendation_ids with the original data for reference
|
|
uploaded_recommendation_ids = [rec.id for rec in session.query(Recommendation).filter(
|
|
Recommendation.property_id == property_id,
|
|
Recommendation.description.in_([rec["description"] for rec in recommendations_to_upload])
|
|
)]
|
|
|
|
# Prepare data for bulk insert for RecommendationMaterials
|
|
recommendation_materials_data = [
|
|
{
|
|
"recommendation_id": recommendation_id,
|
|
"material_id": part["id"],
|
|
"depth": part["depths"][0] if part["depths"] else None,
|
|
"quantity": part["quantity"],
|
|
"quantity_unit": part["quantity_unit"],
|
|
"estimated_cost": part["estimated_cost"],
|
|
}
|
|
for rec, recommendation_id in zip(recommendations_to_upload, uploaded_recommendation_ids)
|
|
for part in rec["parts"]
|
|
]
|
|
|
|
session.bulk_insert_mappings(RecommendationMaterials, recommendation_materials_data)
|
|
|
|
# flush the changes to get the newly created IDs
|
|
session.flush()
|
|
|
|
return uploaded_recommendation_ids
|