Source code for ralph.checkpoint.step_outcome

"""Outcome metadata for checkpoint execution steps."""

from __future__ import annotations

from dataclasses import dataclass, field


[docs] @dataclass(frozen=True) class StepOutcome: """Outcome metadata for a single execution step.""" kind: str output: str | None = None files_modified: list[str] = field(default_factory=list) exit_code: int | None = None recoverable: bool | None = None error: str | None = None completed: str | None = None remaining: str | None = None reason: str | None = None
[docs] @classmethod def success( cls, output: str | None = None, files_modified: list[str] | None = None, ) -> StepOutcome: """Create a success outcome.""" return cls( kind="success", output=output, files_modified=files_modified or [], exit_code=0, )
[docs] @classmethod def failure(cls, error: str, *, recoverable: bool) -> StepOutcome: """Create a failure outcome.""" return cls(kind="failure", error=error, recoverable=recoverable)
[docs] @classmethod def partial(cls, completed: str, remaining: str) -> StepOutcome: """Create a partial outcome.""" return cls(kind="partial", completed=completed, remaining=remaining)
[docs] @classmethod def skipped(cls, reason: str) -> StepOutcome: """Create a skipped outcome.""" return cls(kind="skipped", reason=reason)
[docs] def to_dict(self) -> dict[str, object]: """Return a JSON-safe dictionary representation.""" return { "kind": self.kind, "output": self.output, "files_modified": list(self.files_modified), "exit_code": self.exit_code, "recoverable": self.recoverable, "error": self.error, "completed": self.completed, "remaining": self.remaining, "reason": self.reason, }