mlox.service

Service base abstractions and lifecycle contract for deployable MLOX services.

Purpose:

Provide the common dataclass model and lifecycle methods shared by all concrete services.

Key public classes/functions:

AbstractService core base class with setup/spin/check/teardown interfaces
helper methods for compose execution, dependency lookup, and URL/port registration

Expected runtime mode:

Remote executor backend used from CLI/UI/TUI workflows

Related modules (plain-text links):

logger = <Logger mlox.service (INFO)>

class MloxTemplate(string.Template):

Simple service artifact template using @variable placeholders.

delimiter = '@'

idpattern = '[_a-zA-Z][_a-zA-Z0-9]*'

pattern = re.compile('\n @(?:\n (?P<escaped>@) | # Escape sequence of two delimiters\n (?P<named>[_a-zA-Z][_a-zA-Z0-9]*) | # delimiter and a Python identifier\n , re.IGNORECASE|re.VERBOSE)

class ServiceCapability(enum.StrEnum):

User-facing service capabilities advertised by service configs/classes.

HEALTH = <ServiceCapability.HEALTH: 'health'>

WEB_UI = <ServiceCapability.WEB_UI: 'web_ui'>

SECRET_MANAGER = <ServiceCapability.SECRET_MANAGER: 'secret_manager'>

REPOSITORY = <ServiceCapability.REPOSITORY: 'repository'>

MODEL_REGISTRY = <ServiceCapability.MODEL_REGISTRY: 'model_registry'>

MODEL_SERVER = <ServiceCapability.MODEL_SERVER: 'model_server'>

MONITOR = <ServiceCapability.MONITOR: 'monitor'>

OBSERVABILITY = <ServiceCapability.OBSERVABILITY: 'observability'>

DATA_WAREHOUSE = <ServiceCapability.DATA_WAREHOUSE: 'data_warehouse'>

OBJECT_STORAGE = <ServiceCapability.OBJECT_STORAGE: 'object_storage'>

SPREADSHEET = <ServiceCapability.SPREADSHEET: 'spreadsheet'>

DATABASE = <ServiceCapability.DATABASE: 'database'>

VECTOR_DATABASE = <ServiceCapability.VECTOR_DATABASE: 'vector_database'>

CACHE = <ServiceCapability.CACHE: 'cache'>

MESSAGE_BROKER = <ServiceCapability.MESSAGE_BROKER: 'message_broker'>

WORKFLOW_ORCHESTRATOR = <ServiceCapability.WORKFLOW_ORCHESTRATOR: 'workflow_orchestrator'>

FEATURE_STORE = <ServiceCapability.FEATURE_STORE: 'feature_store'>

CONTAINER_REGISTRY = <ServiceCapability.CONTAINER_REGISTRY: 'container_registry'>

DEPLOYMENT = <ServiceCapability.DEPLOYMENT: 'deployment'>

LLM = <ServiceCapability.LLM: 'llm'>

DASHBOARD = <ServiceCapability.DASHBOARD: 'dashboard'>

DEVELOPER_TOOLS = <ServiceCapability.DEVELOPER_TOOLS: 'developer_tools'>

SERVICE_HEALTHY_STATUSES = {'running'}

SERVICE_HEALTH_STATUSES = {'terminating', 'unknown', 'failed', 'un-initialized', 'starting', 'stopped', 'error', 'running'}

def service_health_payload(service: Any, probe: Optional[Mapping[str, Any]]) -> Dict[str, Any]:

Return a normalized health payload from a service probe result.

class AbstractHealthService(abc.ABC):

Optional service capability for richer live health information.

capabilities: ClassVar[set[ServiceCapability]] = {<ServiceCapability.HEALTH: 'health'>}

@abstractmethod

def get_health(self, conn) -> Dict[str, Any]:

Return a rich live health payload for this service.

class AbstractSecretManagerService(abc.ABC):

Service capability mixin for services that provide a secret manager client.

capabilities: ClassVar[set[ServiceCapability]] = {<ServiceCapability.SECRET_MANAGER: 'secret_manager'>}

@abstractmethod

def get_secret_manager( self, infra: mlox.infra.Infrastructure) -> mlox.secret_manager.AbstractSecretManager:

Return an AbstractSecretManager client for this service.

class AbstractWebUIService(abc.ABC):

Service capability mixin for services with a browser-facing UI.

capabilities: ClassVar[set[ServiceCapability]] = {<ServiceCapability.WEB_UI: 'web_ui'>}

web_ui_url_label: ClassVar[str | None] = None

web_ui_login_fields: ClassVar[tuple[str, ...]] = ()

def get_web_ui_address(self) -> str:

Return the preferred browser URL for this service, if available.

def get_web_ui_login(self, bundle: typing.Any | None = None) -> dict[str, str]:

Return browser-login credentials for this service, if available.

@dataclass

class AbstractRepositoryService(abc.ABC):

Service capability mixin for repository provider services.

capabilities: ClassVar[set[ServiceCapability]] = {<ServiceCapability.REPOSITORY: 'repository'>}

repo_name: str = ''

orchestrator_uuid: str | None = None

created_timestamp: str

modified_timestamp: str

@abstractmethod

def get_url(self) -> str:

@abstractmethod

def git_clone(self, conn) -> None:

@abstractmethod

def git_pull(self, conn) -> None:

def get_repository_root(self) -> str:

Return the absolute repository checkout root path when known.

def repository_summary(self) -> dict[str, typing.Any]:

Return non-IO metadata for repository overview screens.

def get_deploy_keys(self) -> dict[str, str]:

Return deploy keys safe to show or copy in a UI.

def list_repository_tree(self, conn) -> list[dict[str, typing.Any]]:

List repository files for read-only browsing.

def read_repository_file(self, conn, path: str) -> str:

Read one repository file as text.

@dataclass

class AbstractModelRegistryService(abc.ABC):

Service capability mixin for model registry services.

capabilities: ClassVar[set[ServiceCapability]] = {<ServiceCapability.MODEL_REGISTRY: 'model_registry'>}

@abstractmethod

def list_models(self, filter: str | None = None) -> List[Dict[str, Any]]:

def load_artifact( self, model_name: str, model_version: str, artifact_path: str) -> typing.Any | None:

Load a model artifact when supported by the registry implementation.

@dataclass

class AbstractModelServerService(abc.ABC):

Service capability mixin for model-serving services.

capabilities: ClassVar[set[ServiceCapability]] = {<ServiceCapability.MODEL_SERVER: 'model_server'>}

registry_uuid: str | None = None

@abstractmethod

def is_model(self, name: str) -> bool:

@abstractmethod

def get_registry(self) -> AbstractModelRegistryService | None:

def list_supported_models(self) -> List[Dict[str, Any]]:

Return models this endpoint can serve or route to.

def get_example( self, model: Optional[Dict[str, Any]] = None, input_example: typing.Any | None = None) -> str:

Return a concrete invocation example for this model endpoint.

class AbstractMonitorService(abc.ABC):

Service capability mixin for project-level monitoring providers.

capabilities: ClassVar[set[ServiceCapability]] = {<ServiceCapability.MONITOR: 'monitor'>}

@abstractmethod

def get_monitor_snapshot(self, bundle: Any) -> Dict[str, Any]:

Return a compact host/resource monitoring snapshot for one bundle.

@dataclass

class AbstractWorkflowOrchestratorService(abc.ABC):

Service capability mixin for workflow orchestration providers.

capabilities: ClassVar[set[ServiceCapability]] = {<ServiceCapability.WORKFLOW_ORCHESTRATOR: 'workflow_orchestrator'>}

@abstractmethod

def list_workflows(self) -> List[Dict[str, Any]]:

Return workflow/DAG metadata including recent run information.

class ServiceLookup(typing.Protocol):

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as::

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

ServiceLookup(*args, **kwargs)

def get_service_by_uuid(self, service_uuid: str) -> Optional[AbstractService]:

def get_service_by_name(self, service_name: str) -> Optional[AbstractService]:

@dataclass

class AbstractService(abc.ABC):

name: str

service_config_id: str

template: str

target_path: str

uuid: str

target_docker_script: str = 'docker-compose.yaml'

target_docker_env: str = 'service.env'

service_urls: Dict[str, str]

service_ports: Dict[str, int]

compose_service_names: Dict[str, str]

state: Literal['un-initialized', 'running', 'stopped', 'unknown'] = 'un-initialized'

certificate: str = ''

exec: mlox.executors.UbuntuTaskExecutor

def set_task_executor(self, exec: mlox.executors.UbuntuTaskExecutor) -> None:

def bind_service_lookup(self, lookup: ServiceLookup) -> None:

def clear_service_lookup(self) -> None:

def service_dir(self) -> pathlib.Path:

Return the directory containing the concrete service implementation.

def render_template(self, template_name: str, variables: Mapping[str, Any]) -> str:

Render a service-local template with explicit @variable values.

Templates are resolved relative to the concrete service implementation module, so service artifacts can live next to k8s.py, docker.py, and the service mlox.*.yaml descriptor.

def render_template_to_file( self, conn, template_name: str, remote_path: str, variables: Mapping[str, Any]) -> str:

Render a service-local template and write it to a remote path.

@staticmethod

def yaml_scalar(value: Any) -> str:

Return a safe one-line YAML scalar representation.

@staticmethod

def indent_block(value: str, spaces: int) -> str:

Indent a multiline block for embedding in YAML block scalars.

@abstractmethod

def setup(self, conn) -> None:

@abstractmethod

def teardown(self, conn) -> None:

@abstractmethod

def check(self, conn) -> Dict:

@abstractmethod

def get_secrets(self) -> Dict[str, Dict]:

Return a mapping of secret identifiers to structured secret payloads.

def spin_up(self, conn) -> bool:

Start the service.

Concrete services should override this method to perform any provisioning logic required to run the service. The default implementation exists solely to satisfy type checkers and unit tests that rely on instantiating AbstractService subclasses without providing spin control behavior.

def spin_down(self, conn) -> bool:

Stop the service.

def restart(self, conn) -> bool:

Repair or restart the service without changing its configuration.

def compose_up(self, conn) -> bool:

Bring up the docker compose stack for this service.

def compose_restart(self, conn) -> bool:

Reconcile the docker compose stack for this service.

def compose_down(self, conn, *, remove_volumes: bool = False) -> bool:

Tear down the docker compose stack for this service.

def compose_service_status(self, conn) -> Dict[str, str]:

Return docker compose state for tracked services.

Attempts to use docker compose ps to retrieve structured service state information. Falls back to inspecting individual containers when the structured output is unavailable.

def compose_service_log_tail(self, conn, label: str, tail: int = 200) -> str:

Return the recent log tail for a tracked compose service label.

Resolves the compose service name to a container name using the same heuristics as compose_service_status and then returns the last tail lines using the remote helper.

def log_labels(self) -> List[str]:

Return selectable log labels for this service.

Docker-backed services use their compose labels by default. Kubernetes services can override this to expose deployment, pod, or container labels without making callers know the backend.

def service_log_tail(self, conn, label: str | None = None, tail: int = 200) -> str:

Return recent logs for the selected service log label.

def get_dependent_service(self, service_uuid: str) -> Optional[AbstractService]:

Get a dependent service by UUID via the bound service lookup.

def get_dependent_service_by_name(self, service_name: str) -> Optional[AbstractService]:

Get a dependent service by name via the bound service lookup.

def dump_state(self, conn) -> None:

Persist service debugging artifacts to the target directory.