mlox.service API documentation

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:

@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 compose_up(self, conn) -> bool:

Bring up 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 get_dependent_service(self, service_uuid: str) -> Optional[AbstractService]:

Get a dependent service by UUID using singleton registry.

def dump_state(self, conn) -> None:

Persist service debugging artifacts to the target directory.