mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 10:13:58 +00:00
4ee2b259c9
- Add build-context helper to copy Docker context safely (ignore runtime data, keep symlinks) - Update UI and core actions to use ~/.prole/build and shared copy helper - Add/adjust tests and scripts; introduce knoe ops helpers and update manifests Co-authored-by: Junie <junie@jetbrains.com>
90 lines
3.0 KiB
Python
90 lines
3.0 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]
|