Developer
8cdeadd6dd
PR 1 of 4 for the runtime service registry change. - Add Fernet encryption helper (services/secrets.py) with a required MANAGE_ENCRYPTION_KEY; validate it on startup. - Add closed integrations/ registry with Pydantic config + widget-config definitions for grafana, prometheus, jellyfin, nextcloud, and ssh_tasks. - Add services + service_task_runs tables and SettingsStore CRUD with cascade-delete (defensive until widgets carry service_id). - Add /api/services/types and /api/services/instances CRUD (encrypted secrets, secrets_set flags only; never plaintext). - Declare cryptography as a direct dependency. - Require MANAGE_ENCRYPTION_KEY in compose + .env.example + README. - Add 25 backend tests (registry, encryption, CRUD, cascade, task-run history). Verification: ruff clean; pytest 225 passed; frontend lint/build green.
118 lines
3.4 KiB
Python
118 lines
3.4 KiB
Python
"""Base classes for service integrations.
|
|
|
|
A *service definition* is a closed, compile-time description of an external service
|
|
the app can talk to (Grafana, Jellyfin, …). 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 Any
|
|
|
|
from pydantic import BaseModel
|
|
|
|
|
|
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.
|
|
"""
|
|
|
|
|
|
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
|
|
|
|
|
|
@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,
|
|
)
|
|
|
|
|
|
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)
|