Developer
8ff735d644
Slice 4b backend half. Jellyfin and Jellyseerr clients are now resolved from service instances instead of machine-level app config. - Add jellyseerr service definition (6 service types total); add user_id to the Jellyfin service config. - dependencies.py: jellyfin_service_id query param + _service_record (decrypt-on-read); get_jellyfin_client / get_jellyseerr_client / get_user_id resolve against the service registry (first enabled instance as fallback). - SSH/Files transport (get_ssh_client) unchanged; still uses machine_id. - Update service-registry tests for 6 types. Selection model: split params — ?jellyfin_service_id= for Jellyfin/Jellyseerr, ?machine_id= for SSH/Files. Frontend threading follows in the next PR. Verification: backend ruff clean, pytest 222 passed; frontend green (unchanged).
51 lines
2.0 KiB
Python
51 lines
2.0 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.jellyseerr import DEFINITION as JELLYSEERR
|
|
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,
|
|
JELLYSEERR.service_type: JELLYSEERR,
|
|
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
|