core: add render-only service adapters

This commit is contained in:
scawful
2025-12-30 16:14:33 -05:00
parent b543ff989c
commit 7b0c3fd3e9
6 changed files with 110 additions and 44 deletions

View File

@@ -1,5 +1,8 @@
"""AFS service management primitives.""" """AFS service management primitives."""
from .adapters.base import ServiceAdapter
from .adapters.launchd import LaunchdAdapter
from .adapters.systemd import SystemdAdapter
from .manager import ServiceManager from .manager import ServiceManager
from .models import ServiceDefinition, ServiceState, ServiceStatus, ServiceType from .models import ServiceDefinition, ServiceState, ServiceStatus, ServiceType
@@ -9,4 +12,7 @@ __all__ = [
"ServiceState", "ServiceState",
"ServiceStatus", "ServiceStatus",
"ServiceType", "ServiceType",
"ServiceAdapter",
"LaunchdAdapter",
"SystemdAdapter",
] ]

View File

@@ -0,0 +1,18 @@
"""Base classes for AFS service adapters."""
from __future__ import annotations
from dataclasses import dataclass
from ..models import ServiceDefinition
@dataclass
class ServiceAdapter:
platform_name: str
def render(self, definition: ServiceDefinition) -> str:
raise NotImplementedError
def file_extension(self) -> str:
return ""

View File

@@ -0,0 +1,29 @@
"""Launchd adapter (render-only) for AFS services."""
from __future__ import annotations
import json
from .base import ServiceAdapter
from ..models import ServiceDefinition
class LaunchdAdapter(ServiceAdapter):
def __init__(self) -> None:
super().__init__(platform_name="darwin")
def render(self, definition: ServiceDefinition) -> str:
payload: dict[str, object] = {
"Label": f"afs.{definition.name}",
"ProgramArguments": list(definition.command),
"RunAtLoad": bool(definition.run_at_load),
"KeepAlive": bool(definition.keep_alive),
}
if definition.working_directory:
payload["WorkingDirectory"] = str(definition.working_directory)
if definition.environment:
payload["EnvironmentVariables"] = dict(definition.environment)
return json.dumps(payload, indent=2)
def file_extension(self) -> str:
return ".plist"

View File

@@ -0,0 +1,38 @@
"""Systemd adapter (render-only) for AFS services."""
from __future__ import annotations
from .base import ServiceAdapter
from ..models import ServiceDefinition, ServiceType
class SystemdAdapter(ServiceAdapter):
def __init__(self) -> None:
super().__init__(platform_name="linux")
def render(self, definition: ServiceDefinition) -> str:
lines = [
"[Unit]",
f"Description={definition.label}",
"",
"[Service]",
f"ExecStart={' '.join(definition.command)}",
]
if definition.working_directory:
lines.append(f"WorkingDirectory={definition.working_directory}")
if definition.environment:
for key, value in definition.environment.items():
lines.append(f"Environment={key}={value}")
if definition.service_type == ServiceType.ONESHOT:
lines.append("Type=oneshot")
if definition.keep_alive:
lines.append("Restart=on-failure")
lines.extend([
"",
"[Install]",
"WantedBy=default.target",
])
return "\n".join(lines)
def file_extension(self) -> str:
return ".service"

View File

@@ -2,7 +2,6 @@
from __future__ import annotations from __future__ import annotations
import json
import platform import platform
import sys import sys
from pathlib import Path from pathlib import Path
@@ -10,6 +9,9 @@ from typing import Iterable
from ..config import load_config_model from ..config import load_config_model
from ..schema import AFSConfig, ServiceConfig from ..schema import AFSConfig, ServiceConfig
from .adapters.base import ServiceAdapter
from .adapters.launchd import LaunchdAdapter
from .adapters.systemd import SystemdAdapter
from .models import ServiceDefinition, ServiceState, ServiceStatus, ServiceType from .models import ServiceDefinition, ServiceState, ServiceStatus, ServiceType
@@ -26,6 +28,7 @@ class ServiceManager:
self.config = config or load_config_model() self.config = config or load_config_model()
self.service_root = service_root or Path.home() / ".config" / "afs" / "services" self.service_root = service_root or Path.home() / ".config" / "afs" / "services"
self.platform_name = platform_name or platform.system().lower() self.platform_name = platform_name or platform.system().lower()
self._adapter = self._build_adapter(self.platform_name)
def list_definitions(self) -> list[ServiceDefinition]: def list_definitions(self) -> list[ServiceDefinition]:
definitions = self._builtin_definitions() definitions = self._builtin_definitions()
@@ -40,11 +43,7 @@ class ServiceManager:
definition = self.get_definition(name) definition = self.get_definition(name)
if not definition: if not definition:
raise KeyError(f"Unknown service: {name}") raise KeyError(f"Unknown service: {name}")
return self._adapter.render(definition)
if self.platform_name.startswith("darwin"):
payload = render_launchd_plist(definition)
return json.dumps(payload, indent=2)
return render_systemd_unit(definition)
def status(self, name: str) -> ServiceStatus: def status(self, name: str) -> ServiceStatus:
definition = self.get_definition(name) definition = self.get_definition(name)
@@ -117,6 +116,14 @@ class ServiceManager:
), ),
} }
def _build_adapter(self, platform_name: str) -> ServiceAdapter:
platform_name = platform_name.lower()
if platform_name.startswith("darwin") or platform_name.startswith("mac"):
return LaunchdAdapter()
if platform_name.startswith("linux"):
return SystemdAdapter()
return SystemdAdapter()
def _resolve_python_executable(self) -> str: def _resolve_python_executable(self) -> str:
if self.config.general.python_executable: if self.config.general.python_executable:
return str(self.config.general.python_executable) return str(self.config.general.python_executable)
@@ -158,41 +165,3 @@ def _merge_definition(
run_at_load=override.auto_start, run_at_load=override.auto_start,
) )
def render_launchd_plist(definition: ServiceDefinition) -> dict[str, object]:
payload: dict[str, object] = {
"Label": f"afs.{definition.name}",
"ProgramArguments": list(definition.command),
"RunAtLoad": bool(definition.run_at_load),
"KeepAlive": bool(definition.keep_alive),
}
if definition.working_directory:
payload["WorkingDirectory"] = str(definition.working_directory)
if definition.environment:
payload["EnvironmentVariables"] = dict(definition.environment)
return payload
def render_systemd_unit(definition: ServiceDefinition) -> str:
lines = [
"[Unit]",
f"Description={definition.label}",
"",
"[Service]",
f"ExecStart={' '.join(definition.command)}",
]
if definition.working_directory:
lines.append(f"WorkingDirectory={definition.working_directory}")
if definition.environment:
for key, value in definition.environment.items():
lines.append(f"Environment={key}={value}")
if definition.service_type == ServiceType.ONESHOT:
lines.append("Type=oneshot")
if definition.keep_alive:
lines.append("Restart=on-failure")
lines.extend([
"",
"[Install]",
"WantedBy=default.target",
])
return "\n".join(lines)

View File

@@ -26,3 +26,9 @@ def test_service_render_contains_execstart() -> None:
manager = ServiceManager(config=AFSConfig(), platform_name="linux") manager = ServiceManager(config=AFSConfig(), platform_name="linux")
unit = manager.render_unit("orchestrator") unit = manager.render_unit("orchestrator")
assert "ExecStart=" in unit assert "ExecStart=" in unit
def test_service_render_launchd_contains_label() -> None:
manager = ServiceManager(config=AFSConfig(), platform_name="darwin")
payload = manager.render_unit("orchestrator")
assert "\"Label\"" in payload