File size: 2,227 Bytes
d06fa57 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | # Copyright (c) 2024-present, Yumeow. Licensed under the MIT License.
"""Small infrastructure helpers shared by API provider adapters."""
from __future__ import annotations
import time
from typing import Any, Callable, Mapping
class RemoteModelCatalog:
"""Fetch and briefly cache one provider's remote model catalog."""
def __init__(
self,
http_client: Callable[[str], Any],
*,
cache_seconds: float = 300.0,
) -> None:
self._http_client = http_client
self._cache_seconds = cache_seconds
self._models: list[str] | None = None
self._cached_at = 0.0
def get(
self,
url: str,
*,
description: str,
api_key: str | None = None,
params: Mapping[str, str] | None = None,
include: Callable[[dict[str, Any]], bool] | None = None,
) -> list[str]:
"""Return a model list, propagating failures without caching them."""
now = time.monotonic()
if self._models is not None and now - self._cached_at < self._cache_seconds:
return list(self._models)
headers = {"Authorization": f"Bearer {api_key}"} if api_key else None
with self._http_client(url) as client:
response = client.get(
url,
headers=headers,
params=dict(params or {}),
timeout=10.0,
)
response.raise_for_status()
payload = response.json()
records = payload.get("data") if isinstance(payload, dict) else None
if not isinstance(records, list):
raise ValueError(f"{description} response does not contain a data list")
models = sorted(
{
model_id.strip()
for record in records
if isinstance(record, dict)
and (include is None or include(record))
and isinstance((model_id := record.get("id")), str)
and model_id.strip()
}
)
if not models:
raise ValueError(f"{description} returned no model identifiers")
self._models = models
self._cached_at = now
return list(models)
|