"""Release a governed DataFlow as an immutable data production line.""" from __future__ import annotations import re 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 validate_dataflow_spec 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 def _digest(value: Any, label: str) -> str: normalized = str(value or "") if not re.fullmatch(r"[0-9a-f]{64}", normalized): raise ValueError(f"{label} must be a sha256 hex digest") return normalized def _schema_hashes( flow: dict[str, Any], input_schema_hashes: Any, output_schema_hash: Any, ) -> tuple[dict[str, str], str]: if not isinstance(input_schema_hashes, dict): raise ValueError("input_schema_hashes must be an object") if set(input_schema_hashes) != set(flow["input_schema_refs"]): raise ValueError("input_schema_hashes must cover every input schema") inputs = { ref: _digest(input_schema_hashes[ref], f"schema hash for {ref}") for ref in sorted(input_schema_hashes) } return inputs, _digest(output_schema_hash, "output_schema_hash") class ProductionLineReleaseService: def __init__(self, repository): self.repository = repository def release( self, *, dataflow_uid: str, dataflow_spec: dict[str, Any], source_text: str, input_schema_hashes: dict[str, str], output_schema_hash: 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) inputs, output = _schema_hashes( flow, input_schema_hashes, output_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" ) 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, )