"""Base classes for service integrations. A *service definition* is a closed, compile-time description of an external service the app can talk to (Jellyfin, Prometheus, …). Each definition declares: * its non-secret ``config_schema`` (derived from a Pydantic model), * the secret fields it accepts (API keys / tokens), * the widget kinds it can contribute to the dashboard (each with its own Pydantic-derived config schema). Definitions live in :mod:`media_library_viewer_api.integrations` modules and are assembled into the closed :data:`~media_library_viewer_api.integrations.registry.SERVICE_DEFINITIONS` map. There is no runtime plugin loading. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Annotated, Any from pydantic import BaseModel, BeforeValidator, Field def _validate_service_base_url(value: Any) -> str: """Require an absolute http(s) URL for service ``base_url`` fields. Relative hosts (e.g. ``example.com``) break downstream HTTP clients because ``requests`` treats them as relative paths, so we fail fast with a clear error instead of letting the call silently malfunction. """ if not isinstance(value, str): raise ValueError("base_url must be a string starting with http:// or https://") text = value.strip() if not text: raise ValueError("base_url must not be empty") lowered = text.lower() if not (lowered.startswith("http://") or lowered.startswith("https://")): raise ValueError("base_url must start with http:// or https:// (include the schema)") return text #: Shared annotated type for service ``base_url`` fields. applying the validator #: uniformly across every integration so missing schemas are rejected at the #: config boundary with a helpful message. ServiceBaseUrl = Annotated[ str, Field(description="Absolute URL including the http:// or https:// schema."), BeforeValidator(_validate_service_base_url), ] class ServiceConfigBase(BaseModel): """Base for per-service non-secret config models. Subclass this in each integration module and declare the connection fields. The JSON schema is derived via ``model_json_schema()`` and exposed to the UI. Connection URLs should use the :data:`ServiceBaseUrl` type so the ``http(s)://`` schema is enforced consistently across integrations. """ class WidgetConfigBase(BaseModel): """Base for per-widget config models. Subclass this for each widget kind a service provides. Widget configs never hold secrets; credentials live on the parent service record. """ model_config = {"extra": "forbid"} @dataclass(frozen=True) class SecretField: """A secret field stored encrypted on the service record.""" key: str label: str required: bool = False helper: str | None = None @dataclass(frozen=True) class WidgetKind: """A widget kind contributed by a service definition.""" kind: str name: str description: str config_schema: dict[str, Any] default_config: dict[str, Any] = field(default_factory=dict) refresh_interval_ms: int = 0 config_model: type[WidgetConfigBase] | None = None @dataclass(frozen=True) class ServiceDefinition: """Closed description of an external service type.""" service_type: str name: str description: str config_model: type[ServiceConfigBase] secret_fields: list[SecretField] widget_kinds: list[WidgetKind] @property def config_schema(self) -> dict[str, Any]: """JSON schema for the service's non-secret config.""" return self.config_model.model_json_schema() @property def secret_keys(self) -> set[str]: return {sf.key for sf in self.secret_fields} def widget_kind(self, kind: str) -> WidgetKind | None: for wk in self.widget_kinds: if wk.kind == kind: return wk return None def widget_kind( kind: str, name: str, description: str, model_cls: type[WidgetConfigBase], *, default_config: dict[str, Any] | None = None, refresh_interval_ms: int = 0, ) -> WidgetKind: """Build a :class:`WidgetKind` from a Pydantic widget-config model.""" schema = model_cls.model_json_schema() # Strip Pydantic's title noise so the exposed schema stays clean. schema.pop("title", None) return WidgetKind( kind=kind, name=name, description=description, config_schema=schema, default_config=dict(default_config or {}), refresh_interval_ms=refresh_interval_ms, config_model=model_cls, ) def validate_config(model_cls: type[BaseModel], config: dict[str, Any] | None) -> dict[str, Any]: """Validate a config dict against a Pydantic model and return the cleaned dict.""" instance = model_cls.model_validate(config or {}) return instance.model_dump(exclude_none=True)