"""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 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 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" ) 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, )