DataOps-platform/app/core/knowledge/publish.py

from __future__ import annotations


class ChangeSetStateMachine:
    _allowed = {
        "pending": {"diffed", "failed", "rolled_back"},
        "diffed": {"building", "failed", "rolled_back"},
        "building": {"validating", "failed", "rolled_back"},
        "validating": {"canonical_active", "failed", "rolled_back"},
        "canonical_active": {"projecting", "complete", "degraded", "rolled_back"},
        "projecting": {"complete", "degraded", "failed", "rolled_back"},
        "degraded": {"projecting", "complete", "rolled_back"},
        "failed": {"building", "rolled_back"},
        "complete": set(),
        "rolled_back": set(),
    }

    def transition(
        self, current: str, target: str, *, impact_truncated: bool = False
    ) -> str:
        if impact_truncated and target == "canonical_active":
            raise ValueError("truncated impact scope cannot become canonical active")
        if target not in self._allowed.get(current, set()):
            raise ValueError(
                f"invalid knowledge change-set transition: {current} -> {target}"
            )
        return target