mirror of
https://github.com/dredx/prole.git
synced 2026-09-24 18:54:32 +00:00
6b89ea23d7
- Finalized stable, repeatable reset logic for the k3d pipeline. - Refactored installer into modular components: core, milestone, runner, and state. - Introduced new UI abstractions with support for ncurses and Tkinter. - Updated initialization scripts and configurations for CloudNativePG, Kerberos, OpenBao, and Monitoring. - Improved pipeline repair and port-forwarding mechanisms.
85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
"""Installer runner that executes milestones sequentially."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Callable, Iterable
|
|
|
|
from .milestone import Milestone, ValidationResult
|
|
from .state import InstallerState
|
|
|
|
EventCallback = Callable[..., None]
|
|
|
|
|
|
class InstallerRunner:
|
|
"""Executes milestones sequentially and emits lifecycle events."""
|
|
|
|
def __init__(self, milestones: Iterable[Milestone], state: InstallerState | None = None):
|
|
self._milestones = {m.id: m for m in milestones}
|
|
self._order = [m.id for m in milestones]
|
|
self.state = state or InstallerState()
|
|
self._listeners: dict[str, list[EventCallback]] = {
|
|
'on_enter_milestone': [],
|
|
'on_validation_error': [],
|
|
'on_progress': [],
|
|
'on_complete': [],
|
|
}
|
|
|
|
def on(self, event: str, callback: EventCallback) -> None:
|
|
if event not in self._listeners:
|
|
raise ValueError(f"Unknown event: {event}")
|
|
self._listeners[event].append(callback)
|
|
|
|
def emit(self, event: str, *args) -> None:
|
|
for cb in self._listeners.get(event, []):
|
|
cb(*args)
|
|
|
|
def run(self, start_id: str | None = None) -> bool:
|
|
if not self._order:
|
|
self.emit('on_complete', None, self.state)
|
|
return True
|
|
|
|
current_id = start_id or self.state.current_id or self._order[0]
|
|
|
|
while current_id:
|
|
milestone = self._milestones.get(current_id)
|
|
if milestone is None:
|
|
self.state.errors.append(f"Unknown milestone id: {current_id}")
|
|
self.emit('on_validation_error', None, self.state, [self.state.errors[-1]])
|
|
return False
|
|
|
|
self.state.current_id = current_id
|
|
self.state.history.append(current_id)
|
|
self.emit('on_enter_milestone', milestone, self.state)
|
|
|
|
errors = milestone.validate(self.state)
|
|
error_list = _normalize_validation(errors)
|
|
if error_list:
|
|
self.state.errors.extend(error_list)
|
|
self.emit('on_validation_error', milestone, self.state, error_list)
|
|
return False
|
|
|
|
def _progress(message: str, percent: float | None = None) -> None:
|
|
if percent is not None:
|
|
self.state.progress[current_id] = percent
|
|
self.emit('on_progress', milestone, self.state, message, percent)
|
|
|
|
milestone.execute(self.state, _progress)
|
|
self.state.mark_completed(current_id)
|
|
|
|
next_id = milestone.next(self.state)
|
|
if not next_id:
|
|
self.emit('on_complete', milestone, self.state)
|
|
return True
|
|
|
|
current_id = next_id
|
|
|
|
self.emit('on_complete', None, self.state)
|
|
return True
|
|
|
|
|
|
def _normalize_validation(result: ValidationResult) -> list[str]:
|
|
if result is None:
|
|
return []
|
|
if isinstance(result, str):
|
|
return [result]
|
|
return [str(item) for item in result]
|