| 132 | |
| 133 | |
| 134 | class RoutingConfigStore: |
| 135 | def __init__( |
| 136 | self, |
| 137 | storage: RoutingConfigStorage | None = None, |
| 138 | base_config: RoutingConfig | None = None, |
| 139 | ) -> None: |
| 140 | self._storage = storage or FileRoutingConfigStorage() |
| 141 | self._base_config = copy.deepcopy(base_config or DEFAULT_CONFIG) |
| 142 | raw = self._storage.load() |
| 143 | self._overrides = _sanitize_overrides(raw) |
| 144 | self._default_mode = _sanitize_default_mode(raw) |
| 145 | |
| 146 | def config(self) -> RoutingConfig: |
| 147 | cfg: RoutingConfig = copy.deepcopy(self._base_config) |
| 148 | for mode_name, tier_map in self._overrides.items(): |
| 149 | mode = RoutingMode(mode_name) |
| 150 | table = _mode_table(cfg, mode) |
| 151 | for tier_name, payload in tier_map.items(): |
| 152 | tier = Tier(tier_name) |
| 153 | table[tier] = TierConfig( |
| 154 | primary=str(payload["primary"]), |
| 155 | fallback=list(payload.get("fallback", [])), |
| 156 | hard_pin=bool(payload.get("hard_pin", False)), |
| 157 | ) |
| 158 | return cfg |
| 159 | |
| 160 | def export(self) -> dict[str, Any]: |
| 161 | cfg = self.config() |
| 162 | modes: dict[str, dict[str, Any]] = {} |
| 163 | for mode in RoutingMode: |
| 164 | active = _mode_table(cfg, mode) |
| 165 | overridden_tiers = self._overrides.get(mode.value, {}) |
| 166 | tier_rows: dict[str, Any] = {} |
| 167 | for tier in Tier: |
| 168 | tc = active[tier] |
| 169 | tier_rows[tier.value] = { |
| 170 | "primary": tc.primary, |
| 171 | "fallback": list(tc.fallback), |
| 172 | "overridden": tier.value in overridden_tiers, |
| 173 | "hard_pin": tc.hard_pin, |
| 174 | "selection_mode": "hard-pin" if tc.hard_pin else "adaptive", |
| 175 | } |
| 176 | modes[mode.value] = {"tiers": tier_rows} |
| 177 | return { |
| 178 | "source": "local-file", |
| 179 | "editable": True, |
| 180 | "default_mode": self._default_mode.value, |
| 181 | "modes": modes, |
| 182 | } |
| 183 | |
| 184 | def default_mode(self) -> RoutingMode: |
| 185 | return self._default_mode |
| 186 | |
| 187 | def set_default_mode(self, mode: RoutingMode) -> dict[str, Any]: |
| 188 | self._default_mode = mode |
| 189 | self._persist() |
| 190 | return self.export() |
| 191 |
no outgoing calls