75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
"""Shared EAV flattening utility.
|
|
|
|
Every endpoint that returns materials (``material_attribute_values``) or
|
|
line items (``line_item_specs``) with expand uses this single function to
|
|
produce the contract shape:
|
|
|
|
[{"attribute_key", "label", "data_type", "value", "unit"}]
|
|
|
|
value extraction:
|
|
number/integer -> value_number (integer cast for `integer`)
|
|
string/enum -> value_string
|
|
boolean -> value_boolean
|
|
date -> value_date (ISO string)
|
|
|
|
unit resolution:
|
|
prefer row.unit_id_at_time when available (prefer_unit_id_at_time=True),
|
|
else fall back to attribute_definitions.unit_id.
|
|
"""
|
|
from app.models.catalog import AttributeDefinition, MaterialAttributeValue
|
|
from app.models.procurement import LineItemSpec
|
|
|
|
_EAVRow = MaterialAttributeValue | LineItemSpec
|
|
|
|
|
|
def _extract_value(ad: AttributeDefinition, row: _EAVRow):
|
|
data_type = ad.data_type
|
|
if data_type in ("number", "integer"):
|
|
if row.value_number is None:
|
|
return None
|
|
return int(row.value_number) if data_type == "integer" else float(row.value_number)
|
|
if data_type in ("string", "enum"):
|
|
return row.value_string
|
|
if data_type == "boolean":
|
|
return row.value_boolean
|
|
if data_type == "date":
|
|
return row.value_date.isoformat() if row.value_date is not None else None
|
|
return None
|
|
|
|
|
|
def _resolve_unit(row: _EAVRow, ad: AttributeDefinition, unit_map: dict, prefer_unit_id_at_time: bool):
|
|
unit_id = None
|
|
if prefer_unit_id_at_time and hasattr(row, "unit_id_at_time") and row.unit_id_at_time is not None:
|
|
unit_id = row.unit_id_at_time
|
|
elif ad.unit_id is not None:
|
|
unit_id = ad.unit_id
|
|
unit = unit_map.get(unit_id) if unit_id is not None else None
|
|
if unit is None:
|
|
return None
|
|
return {"id": unit.id, "code": unit.code}
|
|
|
|
|
|
def flatten_eav_values(
|
|
rows: list[_EAVRow],
|
|
attr_defs_map: dict,
|
|
unit_map: dict,
|
|
prefer_unit_id_at_time: bool = True,
|
|
include_definition_id: bool = False,
|
|
include_data_type: bool = True,
|
|
) -> list[dict]:
|
|
"""Flatten typed EAV rows into the contract's `value`/`unit` shape."""
|
|
result: list[dict] = []
|
|
for row in rows:
|
|
ad = attr_defs_map[row.attribute_definition_id]
|
|
item: dict = {}
|
|
if include_definition_id:
|
|
item["attribute_definition_id"] = str(row.attribute_definition_id)
|
|
item["attribute_key"] = ad.attribute_key
|
|
item["label"] = ad.label
|
|
if include_data_type:
|
|
item["data_type"] = ad.data_type
|
|
item["value"] = _extract_value(ad, row)
|
|
item["unit"] = _resolve_unit(row, ad, unit_map, prefer_unit_id_at_time)
|
|
result.append(item)
|
|
return result
|