DataOps-platform/app/core/data_rules/release.py

"""Release a governed DataFlow as an immutable data production line."""

from __future__ import annotations

from typing import Any

from app.core.common.identifiers import ensure_governance_uid, new_governance_uid
from app.core.data_rules.compiler import compile_rule_plan
from app.core.data_rules.contracts import read_rule_spec, validate_dataflow_spec
from app.core.data_rules.expressions import validate_rule_expressions
from app.core.data_rules.production_line import resolve_production_line


def _uid(value: Any, label: str) -> str:
    try:
        return ensure_governance_uid({"uid": str(value)})
    except ValueError as exc:
        raise ValueError(f"{label} must be a valid UUIDv7") from exc


def _source(value: Any) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValueError("source_text is required")
    normalized = value.strip()
    if len(normalized) > 20_000:
        raise ValueError("source_text exceeds 20000 characters")
    return normalized


class ProductionLineReleaseService:
    def __init__(self, repository, *, schema_resolver=None):
        self.repository = repository
        self.schema_resolver = schema_resolver

    def release(
        self,
        *,
        dataflow_uid: str,
        dataflow_spec: dict[str, Any],
        source_text: str,
        created_by: str,
    ) -> dict[str, Any]:
        path_uid = _uid(dataflow_uid, "dataflow_uid")
        actor = _uid(created_by, "created_by")
        flow = validate_dataflow_spec(dataflow_spec)
        if flow["dataflow_uid"] != path_uid:
            raise ValueError("dataflow spec uid does not match path dataflow_uid")
        source = _source(source_text)
        if self.schema_resolver is None:
            raise ValueError("trusted schema resolver is not configured")
        input_snapshots = {
            ref: self.schema_resolver.resolve(ref) for ref in flow["input_schema_refs"]
        }
        output_snapshot = self.schema_resolver.resolve(flow["output_schema_ref"])
        inputs = {
            ref: snapshot["schema_hash"]
            for ref, snapshot in sorted(input_snapshots.items())
        }
        output = output_snapshot["schema_hash"]

        standards, rules = self.repository.load_published_assets(flow)
        version = self.repository.begin_dataflow_release(
            dataflow_spec=flow,
            source_text=source,
            input_schema_hashes=inputs,
            output_schema_hash=output,
            created_by=actor,
        )
        version_id = _uid(version.get("id"), "dataflow_version_id")
        compiled: dict[str, dict[str, Any]] = {}
        binding_ids: dict[str, str] = {}

        def add_binding(
            *,
            binding_key: str,
            component_id: str,
            component_kind: str,
            rule_version_id: str,
            stage: str,
            order_no: int,
            idempotency: dict[str, Any] | None,
            provenance: dict[str, str],
        ) -> None:
            rule = rules.get(rule_version_id)
            if not isinstance(rule, dict):
                raise ValueError(
                    f"published rule version {rule_version_id} was not found"
                )
            rule_spec = read_rule_spec(rule.get("rule_spec"))
            rule_snapshot = self.schema_resolver.resolve(
                rule_spec["input_schema_ref"]
            )
            snapshot_fields = {
                field["name"]: field["type"]
                for field in rule_snapshot["fields"]
            }
            validate_rule_expressions(rule_spec, snapshot_fields)
            plan = compiled.setdefault(rule_version_id, compile_rule_plan(rule))
            binding_id = new_governance_uid()
            binding_ids[binding_key] = binding_id
            self.repository.persist_component_plan(
                dataflow_version_id=version_id,
                component_binding_id=binding_id,
                component_id=component_id,
                component_kind=component_kind,
                rule_version_id=rule_version_id,
                stage=stage,
                order_no=order_no,
                idempotency=idempotency,
                provenance=provenance,
                plan=plan,
                schema_hashes={
                    "inputs": inputs,
                    "output": output,
                },
            )

        for component in flow["components"]:
            if component["type"] == "standard.enforce":
                standard_id = component["standard_version_id"]
                standard = standards.get(standard_id)
                if not isinstance(standard, dict):
                    raise ValueError(
                        f"published standard version {standard_id} was not found"
                    )
                for clause_index, clause in enumerate(standard["clauses"]):
                    clause_id = str(clause["clause_id"])
                    binding_key = f"{component['id']}:{clause_id}"
                    add_binding(
                        binding_key=binding_key,
                        component_id=(f"{component['id']}__{clause_id}"[:100]),
                        component_kind="quality.check",
                        rule_version_id=str(clause["rule_version_id"]),
                        stage=component["stage"],
                        order_no=(component["order"] * 1000) + clause_index,
                        idempotency=None,
                        provenance={
                            "standard_version_id": standard_id,
                            "clause_id": clause_id,
                        },
                    )
                continue
            add_binding(
                binding_key=component["id"],
                component_id=component["id"],
                component_kind=component["type"],
                rule_version_id=component["rule_version_id"],
                stage=component["stage"],
                order_no=component["order"] * 1000,
                idempotency=component.get("idempotency"),
                provenance={},
            )

        release_rules = {
            rule_id: {
                **rule,
                "execution_plan": {
                    "backend": compiled[rule_id]["backend"],
                    "plan_hash": compiled[rule_id]["plan_hash"],
                },
            }
            for rule_id, rule in rules.items()
            if rule_id in compiled
        }
        package = resolve_production_line(
            flow,
            standards,
            release_rules,
            component_binding_ids=binding_ids,
        )
        return self.repository.complete_dataflow_release(
            dataflow_version_id=version_id,
            package=package,
        )