DataOps-platform/app/core/orchestration/policy.py

from __future__ import annotations

from typing import Any

from app.core.orchestration.models import PolicyDecision
from app.core.orchestration.spec import (
    validate_schedule_plan,
    validate_workflow_spec,
)


ENVIRONMENTS = {"development", "test", "production"}


def evaluate_candidate(
    workflow_spec: Any,
    schedule_plan: Any,
    *,
    environment: str,
    deploy_disabled: bool,
    canary_required: bool,
) -> PolicyDecision:
    workflow = validate_workflow_spec(workflow_spec)
    validate_schedule_plan(schedule_plan)
    if environment not in ENVIRONMENTS:
        raise ValueError("unsupported environment")

    has_write = any(
        node["type"] == "sql.execute" or node.get("purpose") == "write"
        for node in workflow["nodes"]
    )
    risk = "high" if has_write else ("medium" if environment == "production" else "low")
    reasons = []
    if environment == "production" and not deploy_disabled:
        reasons.append("production candidate must deploy disabled")
    if environment == "production" and has_write and not canary_required:
        reasons.append("production write candidate requires canary")

    return PolicyDecision(
        allowed=not reasons,
        risk=risk,
        reasons=tuple(reasons),
    )