Files
computer-playbook/utils/database_service.py
T
kevinveenbirkenbachandClaude Sonnet 4.6 be6a310d59 refactor(services): implement role-local service registry (req-001)
- Delete group_vars/all/20_services.yml; all service metadata now lives
  in each role's config/main.yml under compose.services.<name>
- Rename generic 'database' service key to explicit 'mariadb'/'postgres'
  across all app roles; remove legacy type: field
- Add utils/service_registry.py and plugins/lookup/service_registry.py
  as the single source of truth for service discovery
- Add utils/database_service.py to centralize DB service resolution
- Delete service_load_enabled filter and service_should_load lookup;
  replaced by service lookup plugin
- Update sys-utils-service-loader to run in constructor stage and use
  generic load_service.yml; remove load_frontend_service.yml
- Add _service_loader_preloading guard to sys-svc-compose,
  sys-svc-webserver-core and sys-ctl-alm-compose to prevent
  health-checks/alarms from firing during preload phase
- Replace manual load_app.yml calls for shared DB/Mailu with assert-
  based preload validation in sys-svc-rdbms and sys-svc-mail
- Update networks.yml.j2 templates to resolve LDAP/AI networks via
  service_registry instead of hardcoded role names
- Migrate all tests to new service discovery API; delete obsolete tests
- Update docs/contributing/code/services.md to reflect role-local model

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-13 19:33:20 +02:00

52 lines
1.4 KiB
Python

from __future__ import annotations
from typing import Any, Dict, Mapping
RDBMS_SERVICE_KEYS = ("mariadb", "postgres")
def _as_mapping(value: Any) -> Dict[str, Any]:
return value if isinstance(value, dict) else {}
def get_compose_services(
applications: Mapping[str, Any],
application_id: str,
) -> Dict[str, Any]:
application = _as_mapping(applications.get(application_id))
compose = _as_mapping(application.get("compose"))
return _as_mapping(compose.get("services"))
def get_database_service_config(
applications: Mapping[str, Any],
application_id: str,
) -> Dict[str, Any]:
service_key = resolve_database_service_key(applications, application_id)
if not service_key:
return {}
return _as_mapping(
get_compose_services(applications, application_id).get(service_key)
)
def resolve_database_service_key(
applications: Mapping[str, Any],
application_id: str,
) -> str:
services = get_compose_services(applications, application_id)
enabled_keys = [
service_key
for service_key in RDBMS_SERVICE_KEYS
if _as_mapping(services.get(service_key)).get("enabled") is True
]
if len(enabled_keys) > 1:
raise ValueError(
f"{application_id}: multiple direct database services are enabled: "
+ ", ".join(enabled_keys)
)
return enabled_keys[0] if enabled_keys else ""