orchestrata_ultimate.promotion.models

Reference › orchestrata_ultimate › promotion › models

The result vocabulary of a deployment. Every stage in the promotion package returns a StageResult, and the runner collects them into a DeployReport. The promoter stores the report in its ledger and prints it.

StageStatus is a string enum with three values: OK, SKIP and FAIL. SKIP is deliberately not a failure and not silence: a re-run of a completed deploy reports a wall of skips, which is how an operator sees that idempotence held. StageResult is a frozen dataclass holding the stage name, its status, a detail string with the measured specifics (which sha, which units, which port refused) and the elapsed seconds. Its failed property is true only for FAIL, and line() renders one fixed-width report line.

DeployReport carries the target environment, the sha that was put on the target, the previous_sha it replaced, the ordered stage results and a rolled_back flag. The sha field is the fact the old promotion ledger lacked. add() appends a result, ok is true when no stage failed, failure returns the first failed stage, as_dict() produces the JSON-ready record (result DEPLOYED or FAILED) and render() produces the human report ending in a verdict line, with the rollback target appended when a rollback happened.

At a glance

Class diagram

Dependency diagram

API

Classes

StageStatus (enum)

Bases: StrEnum · Line 16

Outcome of one stage.

StageResult (dataclass)

Decorators: dataclass(frozen=True) · Line 31

What one stage did, in terms an operator can act on.

DeployReport (dataclass)

Decorators: dataclass · Line 53

The whole run, and the sha it put on the target.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/promotion/models.py

"""Result types for a deployment.

A deploy either happened or it did not, and the record has to say which — with
enough detail that a later reader can tell a genuine no-op from a stage that
was never reached. Hence three statuses rather than a boolean: ``SKIP`` is the
idempotent case (the tree is already at the sha, the lock has not moved) and is
success; ``FAIL`` stops the run.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from enum import StrEnum


class StageStatus(StrEnum):
    """Outcome of one stage.

    ``SKIP`` is deliberately NOT a failure and NOT silence. Re-running a
    completed deploy should report a wall of SKIPs, which is how an operator
    sees that idempotence held rather than guessing at it from the absence of
    output.
    """

    OK = "OK"
    SKIP = "SKIP"
    FAIL = "FAIL"


@dataclass(frozen=True)
class StageResult:
    """What one stage did, in terms an operator can act on.

    ``detail`` carries the measured specifics — which sha, which units, which
    port refused — because "FAIL restart" without them sends the reader back to
    the journal to reconstruct what this code already knew.
    """

    name: str
    status: StageStatus
    detail: str
    seconds: float = 0.0

    @property
    def failed(self) -> bool:
        return self.status is StageStatus.FAIL

    def line(self) -> str:
        return f"[{self.status.value:<4}] {self.name:<22} {self.detail}"


@dataclass
class DeployReport:
    """The whole run, and the sha it put on the target.

    ``sha`` is the field the old ledger lacked, and its absence is why a
    ``PROMOTED`` record could coexist with a 76-commit-stale tree. It is the
    primary fact here, not a decoration.
    """

    env: str
    sha: str = ""
    previous_sha: str = ""
    stages: list[StageResult] = field(default_factory=list)
    rolled_back: bool = False

    def add(self, result: StageResult) -> StageResult:
        self.stages.append(result)
        return result

    @property
    def ok(self) -> bool:
        return not any(s.failed for s in self.stages)

    @property
    def failure(self) -> StageResult | None:
        return next((s for s in self.stages if s.failed), None)

    def as_dict(self) -> dict[str, object]:
        return {
            "env": self.env,
            "sha": self.sha,
            "previous_sha": self.previous_sha,
            "result": "DEPLOYED" if self.ok else "FAILED",
            "rolled_back": self.rolled_back,
            "stages": [
                {
                    "name": s.name,
                    "status": s.status.value,
                    "detail": s.detail,
                    "seconds": round(s.seconds, 3),
                }
                for s in self.stages
            ],
        }

    def render(self) -> str:
        lines = [s.line() for s in self.stages]
        verdict = "DEPLOYED" if self.ok else "FAILED"
        if self.rolled_back:
            verdict += f" (rolled back to {self.previous_sha[:12]})"
        lines.append(f"{verdict} {self.env} @ {self.sha[:12] or '-'}")
        return "\n".join(lines)