Developer
b3b167c075
New service types: - backups: BackupsConfig(ingestion_label), no secrets, summary widget kind. Modeled as a service so it can be named/multi-instanced like others. - authentik: AuthentikConfig(base_url, timeout_seconds), api_token secret (required). Directory source for the upcoming Users tab. Jellyseerr absorption: - JellyfinConfig gains optional jellyseerr_url + jellyseerr_api_key fields. - integrations/jellyseerr.py deleted; registry entry removed. - clients/jellyseerr.py stays (JellyseerrClient still used by enrichment). - One-time idempotent migration in settings_store.ensure_defaults(): jellyseerr service rows merge into a paired Jellyfin (exactly-one merges; multiple picks first unpaired; none/all-paired drops with a logged warning). The api_key is decrypted from secrets before moving to config. Registry is now 8 types: alertmanager, authentik, backups, grafana, jellyfin, nextcloud, prometheus, ssh_tasks. Tests: registry count updated to 8, jellyseerr-absent assertion, new-type definition assertions, and migration tests (single-jellyfin merge, no- jellyfin drop, idempotency). 256 backend tests pass; ruff clean. Refs openspec/changes/services-as-hub-ia/ (spec R6, tasks slice 1).
55 lines
2.3 KiB
Python
55 lines
2.3 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.alertmanager import DEFINITION as ALERTMANAGER
|
|
from media_library_viewer_api.integrations.authentik import DEFINITION as AUTHENTIK
|
|
from media_library_viewer_api.integrations.backups import DEFINITION as BACKUPS
|
|
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,
|
|
ALERTMANAGER.service_type: ALERTMANAGER,
|
|
JELLYFIN.service_type: JELLYFIN,
|
|
NEXTCLOUD.service_type: NEXTCLOUD,
|
|
SSH_TASKS.service_type: SSH_TASKS,
|
|
BACKUPS.service_type: BACKUPS,
|
|
AUTHENTIK.service_type: AUTHENTIK,
|
|
}
|
|
|
|
|
|
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
|