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.
49 lines
1.9 KiB
Python
49 lines
1.9 KiB
Python
"""Closed registry of service definitions.
|
|
|
|
Adding a brand-new service still requires a backend deploy and a module here.
|
|
There is no runtime plugin loading.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from media_library_viewer_api.integrations.base import ServiceDefinition, WidgetKind
|
|
from media_library_viewer_api.integrations.grafana import DEFINITION as GRAFANA
|
|
from media_library_viewer_api.integrations.jellyfin import DEFINITION as JELLYFIN
|
|
from media_library_viewer_api.integrations.nextcloud import DEFINITION as NEXTCLOUD
|
|
from media_library_viewer_api.integrations.prometheus import DEFINITION as PROMETHEUS
|
|
from media_library_viewer_api.integrations.ssh_tasks import DEFINITION as SSH_TASKS
|
|
|
|
SERVICE_DEFINITIONS: dict[str, ServiceDefinition] = {
|
|
GRAFANA.service_type: GRAFANA,
|
|
PROMETHEUS.service_type: PROMETHEUS,
|
|
JELLYFIN.service_type: JELLYFIN,
|
|
NEXTCLOUD.service_type: NEXTCLOUD,
|
|
SSH_TASKS.service_type: SSH_TASKS,
|
|
}
|
|
|
|
|
|
def list_service_types() -> list[str]:
|
|
"""Return all registered service type names (sorted for stable output)."""
|
|
return sorted(SERVICE_DEFINITIONS)
|
|
|
|
|
|
def get_service_definition(service_type: str) -> ServiceDefinition | None:
|
|
"""Return the definition for a service type, or ``None`` if unknown."""
|
|
return SERVICE_DEFINITIONS.get(service_type)
|
|
|
|
|
|
def get_widget_kind(service_type: str, widget_kind: str) -> WidgetKind | None:
|
|
"""Return a widget kind declared by a service definition, or ``None``."""
|
|
definition = get_service_definition(service_type)
|
|
if definition is None:
|
|
return None
|
|
return definition.widget_kind(widget_kind)
|
|
|
|
|
|
def require_service_definition(service_type: str) -> ServiceDefinition:
|
|
"""Return the definition or raise ``ValueError`` for an unknown type."""
|
|
definition = get_service_definition(service_type)
|
|
if definition is None:
|
|
raise ValueError(f"Unknown service type: {service_type}")
|
|
return definition
|