"""Helpers for reading the effective fallback provider chain from config.""" from __future__ import annotations import os from typing import Any def _normalized_base_url(value: Any) -> str: if not isinstance(value, str): return "" return value.strip().rstrip("/") def resolve_entry_api_key(entry: dict[str, Any] | None) -> str | None: """API key for one fallback entry: inline ``api_key``, else ``key_env``. Mirrors the custom-provider convention (``key_env`` names the env var holding the key; ``api_key_env`` accepted as an alias). Returns None when neither yields a non-empty value, letting ``resolve_runtime_provider`` fall through to the provider's standard credential resolution. """ if not isinstance(entry, dict): return None inline = str(entry.get("api_key") or "").strip() if inline: return inline key_env = str(entry.get("key_env") or entry.get("api_key_env") or "").strip() if key_env: return os.getenv(key_env, "").strip() or None return None def _iter_fallback_entries(raw: Any) -> list[dict[str, Any]]: if isinstance(raw, dict): candidates = [raw] elif isinstance(raw, list): candidates = raw else: return [] entries: list[dict[str, Any]] = [] for entry in candidates: if not isinstance(entry, dict): continue provider = str(entry.get("provider") or "").strip() model = str(entry.get("model") or "").strip() if not provider or not model: continue normalized = dict(entry) normalized["provider"] = provider normalized["model"] = model base_url = _normalized_base_url(entry.get("base_url")) if base_url: normalized["base_url"] = base_url entries.append(normalized) return entries def _entry_identity(entry: dict[str, Any]) -> tuple[str, str, str]: return ( str(entry.get("provider") or "").strip().lower(), str(entry.get("model") or "").strip().lower(), _normalized_base_url(entry.get("base_url")).lower(), ) def get_fallback_chain(config: dict[str, Any] | None) -> list[dict[str, Any]]: """Return the effective fallback chain merged across old and new config keys. ``fallback_providers`` remains the primary source of truth and keeps its order. Legacy ``fallback_model`` entries are appended afterwards unless they target the same provider/model/base_url route as an earlier entry. The returned list always contains fresh dict copies. """ config = config or {} chain: list[dict[str, Any]] = [] seen: set[tuple[str, str, str]] = set() for key in ("fallback_providers", "fallback_model"): for entry in _iter_fallback_entries(config.get(key)): identity = _entry_identity(entry) if identity in seen: continue seen.add(identity) chain.append(entry) return chain