Class BaseModel

src/webwright/models/base.py:225–587

Provider-agnostic model backend. Subclasses must override: - class constants ``_API_KEY_FIELD``, ``_ENV_VAR``, ``_LOG_SOURCE``, ``_DEFAULT_CONFIG_CLASS`` (and optionally ``_MAX_RATE_LIMIT_RETRIES``, ``_MAX_TRANSIENT_RETRIES``) - ``_request_headers``, ``_post_url``

class BaseModel:
    """Provider-agnostic model backend.

    Subclasses must override:
      - class constants ``_API_KEY_FIELD``, ``_ENV_VAR``, ``_LOG_SOURCE``,
        ``_DEFAULT_CONFIG_CLASS`` (and optionally ``_MAX_RATE_LIMIT_RETRIES``,
        ``_MAX_TRANSIENT_RETRIES``)
      - ``_request_headers``, ``_post_url``
      - ``_build_payload``, ``_request_metrics_input``
      - ``_extract_text``, ``_usage_metrics_from_payload``

    Optionally override ``_rate_limit_backoff`` / ``_transient_backoff`` for
    custom retry timing.
    """

    _API_KEY_FIELD: str = ""
    _ENV_VAR: str = ""
    _LOG_SOURCE: str = ""
    _MAX_RATE_LIMIT_RETRIES: int = 5
    _MAX_TRANSIENT_RETRIES: int = 5
    _DEFAULT_CONFIG_CLASS: type = BaseModelConfig

    def __init__(self, *, config_class: type | None = None, **kwargs):
        self.config = (config_class or self._DEFAULT_CONFIG_CLASS)(**kwargs)
        self._last_request_metrics: dict[str, int] = {k: 0 for k in _REQUEST_METRIC_KEYS}
        self._last_usage_metrics: dict[str, int] = {k: 0 for k in _USAGE_METRIC_KEYS}
        self._cumulative_request_metrics: dict[str, int] = dict(self._last_request_metrics)
        self._cumulative_usage_metrics: dict[str, int] = dict(self._last_usage_metrics)

        if self._API_KEY_FIELD:
            if not getattr(self.config, self._API_KEY_FIELD, ""):
                setattr(self.config, self._API_KEY_FIELD, os.environ.get(self._ENV_VAR, ""))
            if not getattr(self.config, self._API_KEY_FIELD, ""):
                raise RuntimeError(f"Missing {self._ENV_VAR}.")

    # ---- subclass extension points ------------------------------------------------

    def _request_headers(self) -> dict[str, str]:
        raise NotImplementedError

    def _post_url(self) -> str:
        raise NotImplementedError

    def _build_payload(self, messages: list[dict[str, Any]]) -> dict[str, Any]:
        raise NotImplementedError

    def _build_text_payload(self, messages: list[dict[str, Any]]) -> dict[str, Any]:
        return self._build_payload(messages)

    def _request_metrics_input(self, payload: dict[str, Any]) -> list[dict[str, Any]]:
        raise NotImplementedError

    def _extract_text(self, payload: dict[str, Any]) -> str:
        raise NotImplementedError

    def _usage_metrics_from_payload(self, payload: dict[str, Any]) -> dict[str, int]:
        raise NotImplementedError