Developer
7d49df3e7d
Slice 1 of observability-service-registry. Alertmanager becomes a
first-class service-registry type, mirroring grafana/prometheus.
- integrations/alertmanager.py (new): AlertmanagerConfig
(base_url, timeout_seconds), AlertmanagerAlertsWidgetConfig (optional
severity_filter), shared summarize_alerts() helper, and DEFINITION
(service_type "alertmanager", secret api_key, widget "active_alerts").
- integrations/registry.py: register ALERTMANAGER (7 types now).
- widgets/sources.py: AlertmanagerWidgetSource fetches
{base_url}/api/v1/alerts, sends optional Bearer token from the api_key
secret, applies optional severity_filter, and summarizes via the shared
helper; registered in SERVICE_ADAPTERS.
- routers/monitoring.py: _summary_from_alerts delegates to the shared
summarize_alerts (behavior unchanged).
- tests: registry now 7 types; /api/services/types lists alertmanager;
4 new adapter tests (summarize, severity filter, bearer token, missing
service).
Backend-only slice; the frontend active_alerts widget binding lands in a
later slice. ruff clean; 228 backend tests pass.
Reviewed fresh-context (read-only): no blockers.
53 lines
2.1 KiB
Python
53 lines
2.1 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.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,
|
|
ALERTMANAGER.service_type: ALERTMANAGER,
|
|
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
|