691d78ff06
Extract the duplicated _resolve_service_record helper (identical in routers/monitoring.py and routers/authentik_users.py) into a shared services/service_resolution.py module. Both routers now import resolve_service_record from the shared module. The authentik router previously hardcoded service_type='authentik' in its local copy; the shared helper takes service_type as a param (same as monitoring's did). Tests updated: test_api.py patches now target the correct module paths (resolve_service_record on the monitoring module where it's imported, build_service_record on the service_resolution module). 283 backend tests pass; ruff clean.
38 lines
1.4 KiB
Python
38 lines
1.4 KiB
Python
"""Shared helpers for resolving service instances at request time.
|
|
|
|
Extracted from the duplicated ``_resolve_service_record`` helpers that lived
|
|
in ``routers/monitoring.py`` and ``routers/authentik_users.py``. Both routers
|
|
need the same logic: return the requested service instance (by id), or fall
|
|
back to the first enabled instance of the type. Returns ``None`` when the
|
|
instance does not exist, is the wrong type, is disabled, or when no enabled
|
|
instance of the type is configured.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from media_library_viewer_api.services.settings_store import SettingsStore
|
|
from media_library_viewer_api.widgets.sources import ServiceRecord, build_service_record
|
|
|
|
|
|
def resolve_service_record(
|
|
store: SettingsStore,
|
|
service_type: str,
|
|
service_id: str | None = None,
|
|
) -> ServiceRecord | None:
|
|
"""Return the requested service instance, else the first enabled one.
|
|
|
|
Returns ``None`` when the instance does not exist / is the wrong type, or
|
|
when no enabled instance of ``service_type`` is configured.
|
|
"""
|
|
if service_id:
|
|
row = store.get_service(service_id)
|
|
if not row or row.get("service_type") != service_type:
|
|
return None
|
|
if not row.get("enabled", True):
|
|
return None
|
|
return build_service_record(store, row)
|
|
for row in store.list_services(service_type):
|
|
if row.get("enabled", True):
|
|
return build_service_record(store, row)
|
|
return None
|