Decode raw ClickHouse rows into exposure counts and per-metric stats. Columns are located by name, so a missing one raises KeyError rather than silently reading a neighbour's value.
(
self, rows: list[Any], column_names: Sequence[str]
)
| 153 | params[f"metric_{slot.index}_event"] = slot.spec.event |
| 154 | |
| 155 | def decode_rows( |
| 156 | self, rows: list[Any], column_names: Sequence[str] |
| 157 | ) -> tuple[dict[str, int], dict[int, dict[str, VariantStats]]]: |
| 158 | """Decode raw ClickHouse rows into exposure counts and per-metric stats. |
| 159 | |
| 160 | Columns are located by name, so a missing one raises KeyError rather than |
| 161 | silently reading a neighbour's value. |
| 162 | """ |
| 163 | index = {name: position for position, name in enumerate(column_names)} |
| 164 | exposure_counts: dict[str, int] = {} |
| 165 | metric_stats: dict[int, dict[str, VariantStats]] = { |
| 166 | slot.spec.metric_id: {} for slot in self._slots |
| 167 | } |
| 168 | for row in rows: |
| 169 | variant = str(row[index["variant"]]) |
| 170 | n = int(row[index["n"]]) |
| 171 | exposure_counts[variant] = n |
| 172 | for slot in self._slots: |
| 173 | metric_stats[slot.spec.metric_id][variant] = slot.decode(n, row, index) |
| 174 | return exposure_counts, metric_stats |
no test coverage detected