"""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) rule_backend_support: dict[str, frozenset[str]] = {} def validate_published_rule(rule_version_id: str) -> None: if rule_version_id in rule_backend_support: return 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"] } supported = validate_rule_expressions( rule_spec, snapshot_fields ) if not supported: raise ValueError("rule has no supported execution backend") rule_backend_support[rule_version_id] = supported # Validate every loaded rule before allocating any release version. # The repository is expected to scope this mapping to published assets # available to the release, so no loaded expression is left unchecked. for loaded_rule_version_id in rules: validate_published_rule(loaded_rule_version_id) # Ensure referenced standards and their clauses resolve to loaded, # already-validated rule versions before beginning the release. for component in flow["components"]: if component["type"] == "standard.enforce": standard = standards.get(component["standard_version_id"]) if not isinstance(standard, dict): raise ValueError( "published standard version " f"{component['standard_version_id']} was not found" ) for clause in standard["clauses"]: validate_published_rule(str(clause["rule_version_id"])) else: validate_published_rule(component["rule_version_id"]) 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" ) if rule_version_id not in compiled: compiled[rule_version_id] = compile_rule_plan( rule, supported_backends=rule_backend_support[rule_version_id], ) plan = compiled[rule_version_id] 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, )