Coverage for backend/django/core/auxiliary/services/result_series_identity.py: 92%
64 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from dataclasses import dataclass
2import json
3from typing import Any
5from core.auxiliary.enums.unitsLibrary import get_unit_choices
6from core.auxiliary.models.IndexedItem import IndexedItem
7from core.auxiliary.models.PropertyValue import PropertyValue
8from pydantic import BaseModel, ConfigDict
11@dataclass(frozen=True)
12class ResultSeriesIndexedItemIdentity:
13 """Stable metadata for one index attached to a result property value."""
15 id: int
16 key: str | None
17 label: str
18 type: str
21@dataclass(frozen=True)
22class ResultSeriesIdentity:
23 """Shared identity for result graph, summary, and object table series."""
25 key: str
26 label: str
27 unit: str | None
28 property_value_id: int
29 indexed_items: tuple[ResultSeriesIndexedItemIdentity, ...]
32class ResultSeriesIndexedItemPayload(BaseModel):
33 model_config = ConfigDict(extra="forbid")
35 id: int
36 key: str | None
37 label: str
38 type: str
41class ResultSeriesIdentityPayload(BaseModel):
42 model_config = ConfigDict(extra="forbid")
44 key: str
45 label: str
46 unit: str | None
47 property_value_id: int
48 indexed_items: list[ResultSeriesIndexedItemPayload]
51def result_series_identity(property_value: PropertyValue) -> ResultSeriesIdentity:
52 """Return the backend-authoritative identity for one result property value."""
54 indexed_items = tuple(
55 ResultSeriesIndexedItemIdentity(
56 id=indexed_item.pk,
57 key=_stable_indexed_item_key(indexed_item.key),
58 label=indexed_item.displayName,
59 type=indexed_item.type,
60 )
61 for indexed_item in _ordered_indexed_items(property_value)
62 )
63 property_info = property_value.property
64 index_labels = [item.label for item in indexed_items if item.label]
65 label_parts = [property_info.displayName, *index_labels]
66 return ResultSeriesIdentity(
67 key=f"property_value:{property_value.pk}",
68 label=" ".join(part for part in label_parts if part),
69 unit=_property_value_unit(property_value),
70 property_value_id=property_value.pk,
71 indexed_items=indexed_items,
72 )
75def _ordered_indexed_items(property_value: PropertyValue) -> list[IndexedItem]:
76 """Return schema-ordered indexes, preserving custom indexed values if present."""
78 fallback_indexes = list(property_value.indexedItems.all())
79 try:
80 ordered_indexes = property_value.get_indexed_items()
81 except Exception:
82 return fallback_indexes
83 return ordered_indexes or fallback_indexes
86def result_series_identity_payload(
87 property_value: PropertyValue,
88) -> ResultSeriesIdentityPayload:
89 """Return the API/cache payload form of the shared result identity."""
91 identity = result_series_identity(property_value)
92 return ResultSeriesIdentityPayload(
93 key=identity.key,
94 label=identity.label,
95 unit=identity.unit,
96 property_value_id=identity.property_value_id,
97 indexed_items=[
98 ResultSeriesIndexedItemPayload(
99 id=indexed_item.id,
100 key=indexed_item.key,
101 label=indexed_item.label,
102 type=indexed_item.type,
103 )
104 for indexed_item in identity.indexed_items
105 ],
106 )
109def _stable_indexed_item_key(value: Any) -> str | None:
110 if value is None: 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true
111 return None
112 if isinstance(value, str): 112 ↛ 114line 112 didn't jump to line 114 because the condition on line 112 was always true
113 return value
114 return json.dumps(value, sort_keys=True, separators=(",", ":"))
117def _property_value_unit(property_value: PropertyValue) -> str | None:
118 property_info = property_value.property
119 unit_value = property_info.unit
120 if not unit_value:
121 return None
122 for raw_unit, display_label in get_unit_choices(property_info.unitType):
123 if raw_unit == unit_value:
124 return display_label
125 return unit_value