Developer
e7bd0afdd1
ServiceDataHarness (services/service_data.py): lifecycle-only registry of per-concern storage — DB provisioning, idempotent migrations (ALTER TABLE duplicate-column-name caught per-statement), cascade_delete(service_id). QbittorrentSampleStore: append/window/prune (MAX_SAMPLES=120) in qbittorrent.db. QbittorrentClient: cookie-login Web API client (403 re-login, /sync/maindata). Integration registered with 3 widget kinds (totals/active/ speed). Harness initialized in main.py lifespan. Backend: 314 pytest pass (21 new), ruff clean.
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.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.qbittorrent import DEFINITION as QBITTORRENT
|
|
from media_library_viewer_api.integrations.ssh_tasks import DEFINITION as SSH_TASKS
|
|
|
|
SERVICE_DEFINITIONS: dict[str, ServiceDefinition] = {
|
|
PROMETHEUS.service_type: PROMETHEUS,
|
|
ALERTMANAGER.service_type: ALERTMANAGER,
|
|
JELLYFIN.service_type: JELLYFIN,
|
|
NEXTCLOUD.service_type: NEXTCLOUD,
|
|
QBITTORRENT.service_type: QBITTORRENT,
|
|
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
|