mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
53d9f21f73
The ASHP bundle is priced from the rate sheet (ADR-0025); the catalogue row is read only for its material id, which is nullable end-to-end. The live `material` catalogue has no `air_source_heat_pump` row, so `products.get` raised `ValueError: no active product` and aborted every ASHP-eligible property. Add `ProductNotFound(ValueError)` + a concrete `ProductRepository .get_optional`, raise the typed error from both repos, and have `_ashp_option` look the row up optionally — a missing row now yields an ASHP Option with `material_id=None` rather than crashing. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
60 lines
2.6 KiB
Python
60 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
from sqlmodel import Session, col, select
|
|
|
|
from domain.modelling.contingencies import contingency_rate
|
|
from domain.modelling.product import Product
|
|
from infrastructure.postgres.product_table import MaterialRow
|
|
from repositories.product.product_repository import ProductNotFound, ProductRepository
|
|
|
|
|
|
# The domain ``MeasureType`` vocabulary and the catalogue's ``material.type``
|
|
# pgEnum drifted apart: these five measures are spelled differently on the
|
|
# catalogue side (and querying the domain spelling raises a pgEnum DataError
|
|
# that poisons the session's transaction). Translate them to the catalogue's
|
|
# own vocabulary at this boundary so the domain enum stays stable. Every other
|
|
# MeasureType already matches its material.type and maps to itself.
|
|
_MATERIAL_TYPE_BY_MEASURE: dict[str, str] = {
|
|
"low_energy_lighting": "low_energy_lighting_installation",
|
|
"gas_boiler_upgrade": "boiler_upgrade",
|
|
"system_tune_up": "roomstat_programmer_trvs",
|
|
"system_tune_up_zoned": "time_temperature_zone_control",
|
|
"sloping_ceiling_insulation": "room_roof_insulation",
|
|
}
|
|
|
|
|
|
class ProductPostgresRepository(ProductRepository):
|
|
"""Reads the ``material`` catalogue table and maps an active row to a
|
|
Product: `total_cost` becomes the fully-loaded `unit_cost_per_m2`, and the
|
|
per-Measure-Type contingency is joined from config."""
|
|
|
|
def __init__(self, session: Session) -> None:
|
|
self._session = session
|
|
|
|
def get(self, measure_type: str) -> Product:
|
|
# Resolve the domain MeasureType to the catalogue's ``material.type``
|
|
# spelling (identity for all but the five drifted types above).
|
|
catalogue_type = _MATERIAL_TYPE_BY_MEASURE.get(measure_type, measure_type)
|
|
# The live catalogue holds many active rows per type; order by id so the
|
|
# pick is deterministic (a re-seed prices the same) rather than relying
|
|
# on the database's physical row order.
|
|
row: MaterialRow | None = self._session.exec(
|
|
select(MaterialRow)
|
|
.where(
|
|
col(MaterialRow.type) == catalogue_type,
|
|
col(MaterialRow.is_active).is_(True),
|
|
)
|
|
.order_by(col(MaterialRow.id))
|
|
).first()
|
|
if row is None:
|
|
raise ProductNotFound(
|
|
f"no active product for measure type {measure_type!r}"
|
|
)
|
|
if row.total_cost is None:
|
|
raise ValueError(f"product {measure_type!r} has no total_cost")
|
|
return Product(
|
|
measure_type=measure_type,
|
|
unit_cost_per_m2=row.total_cost,
|
|
contingency_rate=contingency_rate(measure_type),
|
|
id=row.id,
|
|
)
|