|
|
@@ -0,0 +1,703 @@
|
|
|
+"""Active metadata discovery, incremental change, lineage and correction contracts."""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import hashlib
|
|
|
+import json
|
|
|
+import re
|
|
|
+import uuid
|
|
|
+from collections import Counter
|
|
|
+from copy import deepcopy
|
|
|
+from datetime import UTC, datetime
|
|
|
+from typing import Any
|
|
|
+
|
|
|
+import sqlglot
|
|
|
+from sqlglot import exp
|
|
|
+
|
|
|
+from app.core.common.identifiers import new_governance_uid
|
|
|
+
|
|
|
+SOURCE_KINDS = frozenset({"database", "file", "api"})
|
|
|
+SCHEDULE_TYPES = frozenset({"manual", "interval", "cron"})
|
|
|
+DISCOVERY_MODES = frozenset({"snapshot", "cursor"})
|
|
|
+HEALTH_SIGNAL_TYPES = frozenset({"quality", "freshness", "task_failure", "usage"})
|
|
|
+HEALTH_STATUSES = frozenset({"healthy", "warning", "critical", "unknown"})
|
|
|
+SECRET_MARKERS = ("password", "secret", "token", "credential", "api_key", "private_key")
|
|
|
+INTERVAL_RE = re.compile(r"^PT(?:[1-9]\d*[HMS])+$")
|
|
|
+CRON_PART_RE = re.compile(r"^[\d*/?,LW#-]+$")
|
|
|
+
|
|
|
+
|
|
|
+class ActiveMetadataError(ValueError):
|
|
|
+ pass
|
|
|
+
|
|
|
+
|
|
|
+class ActiveMetadataNotFound(LookupError):
|
|
|
+ pass
|
|
|
+
|
|
|
+
|
|
|
+class ActiveMetadataConflict(RuntimeError):
|
|
|
+ pass
|
|
|
+
|
|
|
+
|
|
|
+def _canonical(value: Any) -> str:
|
|
|
+ return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
|
+
|
|
|
+
|
|
|
+def _hash(value: Any) -> str:
|
|
|
+ return hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest()
|
|
|
+
|
|
|
+
|
|
|
+def _reject_secrets(value: Any, path: str = "payload") -> None:
|
|
|
+ if isinstance(value, dict):
|
|
|
+ for key, nested in value.items():
|
|
|
+ normalized = str(key).lower().replace("-", "_")
|
|
|
+ if any(marker in normalized for marker in SECRET_MARKERS):
|
|
|
+ raise ActiveMetadataError(f"secret-bearing field is not allowed at {path}.{key}")
|
|
|
+ _reject_secrets(nested, f"{path}.{key}")
|
|
|
+ elif isinstance(value, list):
|
|
|
+ for index, nested in enumerate(value):
|
|
|
+ _reject_secrets(nested, f"{path}[{index}]")
|
|
|
+
|
|
|
+
|
|
|
+def _text(value: Any, field: str, limit: int = 300) -> str:
|
|
|
+ result = str(value or "").strip()
|
|
|
+ if not result or len(result) > limit:
|
|
|
+ raise ActiveMetadataError(f"{field} must be between 1 and {limit} characters")
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+def _uuid(value: Any, field: str) -> str:
|
|
|
+ try:
|
|
|
+ return str(uuid.UUID(str(value)))
|
|
|
+ except (TypeError, ValueError, AttributeError) as exc:
|
|
|
+ raise ActiveMetadataError(f"{field} must be a UUID") from exc
|
|
|
+
|
|
|
+
|
|
|
+def _schedule(payload: dict[str, Any]) -> tuple[str, str | None]:
|
|
|
+ schedule_type = str(payload.get("schedule_type") or "manual").strip().lower()
|
|
|
+ if schedule_type not in SCHEDULE_TYPES:
|
|
|
+ raise ActiveMetadataError("schedule_type is unsupported")
|
|
|
+ expression = payload.get("schedule_expression")
|
|
|
+ expression = str(expression).strip() if expression is not None else None
|
|
|
+ if schedule_type == "manual":
|
|
|
+ if expression:
|
|
|
+ raise ActiveMetadataError("manual schedule cannot have schedule_expression")
|
|
|
+ return schedule_type, None
|
|
|
+ if schedule_type == "interval":
|
|
|
+ if not expression or not INTERVAL_RE.fullmatch(expression):
|
|
|
+ raise ActiveMetadataError("interval schedule_expression must be ISO-8601 PT duration")
|
|
|
+ return schedule_type, expression
|
|
|
+ parts = expression.split() if expression else []
|
|
|
+ if len(parts) != 5 or any(not CRON_PART_RE.fullmatch(part) for part in parts):
|
|
|
+ raise ActiveMetadataError("cron schedule_expression must contain five valid fields")
|
|
|
+ return schedule_type, expression
|
|
|
+
|
|
|
+
|
|
|
+def _table_name(table: exp.Table) -> str:
|
|
|
+ return ".".join(part for part in (table.catalog, table.db, table.name) if part)
|
|
|
+
|
|
|
+
|
|
|
+def parse_sql_field_lineage(sql: str, *, dialect: str | None = None) -> dict[str, Any]:
|
|
|
+ """Parse conservative field lineage and preserve unsupported SQL as evidence."""
|
|
|
+ evidence = str(sql or "").strip()
|
|
|
+ if not evidence:
|
|
|
+ return {"status": "failed", "edges": [], "failure_reason": "SQL is empty"}
|
|
|
+ try:
|
|
|
+ statement = sqlglot.parse_one(evidence, read=dialect)
|
|
|
+ if not isinstance(statement, exp.Insert):
|
|
|
+ raise ActiveMetadataError("only INSERT ... SELECT lineage is supported")
|
|
|
+ target_schema = statement.this
|
|
|
+ if not isinstance(target_schema, exp.Schema) or not isinstance(
|
|
|
+ target_schema.this, exp.Table
|
|
|
+ ):
|
|
|
+ raise ActiveMetadataError("INSERT target columns are required")
|
|
|
+ select = statement.expression
|
|
|
+ if not isinstance(select, exp.Select):
|
|
|
+ raise ActiveMetadataError("INSERT source must be a SELECT")
|
|
|
+ targets = [item.name for item in target_schema.expressions]
|
|
|
+ if len(targets) != len(select.expressions):
|
|
|
+ raise ActiveMetadataError("target and select field counts differ")
|
|
|
+ tables = list(select.find_all(exp.Table))
|
|
|
+ aliases = {}
|
|
|
+ for table in tables:
|
|
|
+ name = _table_name(table)
|
|
|
+ aliases[table.alias_or_name] = name
|
|
|
+ aliases[table.name] = name
|
|
|
+ edges = []
|
|
|
+ for target_field, expression in zip(targets, select.expressions, strict=True):
|
|
|
+ columns = list(expression.find_all(exp.Column))
|
|
|
+ if not columns and isinstance(expression, exp.Column):
|
|
|
+ columns = [expression]
|
|
|
+ for column in columns:
|
|
|
+ if column.table:
|
|
|
+ source_asset = aliases.get(column.table)
|
|
|
+ else:
|
|
|
+ source_asset = _table_name(tables[0]) if len(tables) == 1 else None
|
|
|
+ if not source_asset:
|
|
|
+ raise ActiveMetadataError(
|
|
|
+ f"cannot resolve source table for field {column.name}"
|
|
|
+ )
|
|
|
+ edges.append(
|
|
|
+ {
|
|
|
+ "source_asset": source_asset,
|
|
|
+ "source_field": column.name,
|
|
|
+ "target_asset": _table_name(target_schema.this),
|
|
|
+ "target_field": target_field,
|
|
|
+ "relation_type": "derived_from",
|
|
|
+ "evidence": {"kind": "sql", "statement_hash": _hash(evidence)},
|
|
|
+ }
|
|
|
+ )
|
|
|
+ if not edges:
|
|
|
+ raise ActiveMetadataError("no field lineage was resolved")
|
|
|
+ return {"status": "resolved", "edges": edges, "failure_reason": None}
|
|
|
+ except Exception as exc:
|
|
|
+ return {
|
|
|
+ "status": "failed",
|
|
|
+ "edges": [],
|
|
|
+ "failure_reason": str(exc)[:500],
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def _normalize_fields(asset_key: str, fields: Any) -> list[dict[str, Any]]:
|
|
|
+ if fields is None:
|
|
|
+ fields = []
|
|
|
+ if not isinstance(fields, list) or len(fields) > 5000:
|
|
|
+ raise ActiveMetadataError(f"{asset_key}.fields must be a bounded list")
|
|
|
+ result = []
|
|
|
+ seen = set()
|
|
|
+ for index, field in enumerate(fields, 1):
|
|
|
+ if not isinstance(field, dict):
|
|
|
+ raise ActiveMetadataError(f"{asset_key}.fields items must be objects")
|
|
|
+ name = _text(field.get("name"), f"{asset_key}.field.name", 200)
|
|
|
+ if name in seen:
|
|
|
+ raise ActiveMetadataError(f"duplicate field {asset_key}.{name}")
|
|
|
+ seen.add(name)
|
|
|
+ result.append(
|
|
|
+ {
|
|
|
+ "name": name,
|
|
|
+ "data_type": _text(
|
|
|
+ field.get("data_type") or "unknown",
|
|
|
+ f"{asset_key}.{name}.data_type",
|
|
|
+ 100,
|
|
|
+ ),
|
|
|
+ "nullable": bool(field.get("nullable", True)),
|
|
|
+ "ordinal_position": int(field.get("ordinal_position") or index),
|
|
|
+ "default": field.get("default"),
|
|
|
+ "comment": field.get("comment"),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ return sorted(result, key=lambda item: (item["ordinal_position"], item["name"]))
|
|
|
+
|
|
|
+
|
|
|
+def normalize_snapshot(source_uid: str, snapshot: Any) -> list[dict[str, Any]]:
|
|
|
+ if not isinstance(snapshot, dict) or not isinstance(snapshot.get("assets"), list):
|
|
|
+ raise ActiveMetadataError("snapshot.assets must be a list")
|
|
|
+ if len(snapshot["assets"]) > 10000:
|
|
|
+ raise ActiveMetadataError("snapshot contains too many assets")
|
|
|
+ result = []
|
|
|
+ seen = set()
|
|
|
+ for item in snapshot["assets"]:
|
|
|
+ if not isinstance(item, dict):
|
|
|
+ raise ActiveMetadataError("snapshot asset must be an object")
|
|
|
+ namespace = _text(
|
|
|
+ item.get("namespace") or item.get("schema") or "default",
|
|
|
+ "asset.namespace",
|
|
|
+ 200,
|
|
|
+ )
|
|
|
+ name = _text(item.get("name"), "asset.name", 200)
|
|
|
+ key = str(item.get("key") or f"{source_uid}:{namespace}.{name}").strip()
|
|
|
+ if not key.startswith(f"{source_uid}:"):
|
|
|
+ raise ActiveMetadataError("asset key must be scoped to source_uid")
|
|
|
+ if key in seen:
|
|
|
+ raise ActiveMetadataError(f"duplicate asset key {key}")
|
|
|
+ seen.add(key)
|
|
|
+ normalized = {
|
|
|
+ "key": key,
|
|
|
+ "namespace": namespace,
|
|
|
+ "name": name,
|
|
|
+ "asset_type": _text(
|
|
|
+ item.get("asset_type") or "resource", "asset.asset_type", 40
|
|
|
+ ),
|
|
|
+ "comment": item.get("comment"),
|
|
|
+ "fields": _normalize_fields(key, item.get("fields")),
|
|
|
+ }
|
|
|
+ normalized["content_hash"] = _hash(normalized)
|
|
|
+ result.append(normalized)
|
|
|
+ return sorted(result, key=lambda item: item["key"])
|
|
|
+
|
|
|
+
|
|
|
+def _stable_asset_uid(source_uid: str, asset_key: str) -> str:
|
|
|
+ return str(
|
|
|
+ uuid.uuid5(
|
|
|
+ uuid.NAMESPACE_URL,
|
|
|
+ f"dataops-active-metadata:{source_uid}:{asset_key}",
|
|
|
+ )
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+class ActiveMetadataService:
|
|
|
+ def __init__(
|
|
|
+ self,
|
|
|
+ repository,
|
|
|
+ *,
|
|
|
+ uid_factory=new_governance_uid,
|
|
|
+ now_factory=lambda: datetime.now(UTC),
|
|
|
+ ):
|
|
|
+ self.repository = repository
|
|
|
+ self.uid_factory = uid_factory
|
|
|
+ self.now_factory = now_factory
|
|
|
+
|
|
|
+ def create_plan(self, payload: Any, *, actor_uid: str):
|
|
|
+ if not isinstance(payload, dict):
|
|
|
+ raise ActiveMetadataError("plan payload must be an object")
|
|
|
+ _reject_secrets(payload)
|
|
|
+ source_kind = str(payload.get("source_kind") or "").strip().lower()
|
|
|
+ if source_kind not in SOURCE_KINDS:
|
|
|
+ raise ActiveMetadataError("source_kind is unsupported")
|
|
|
+ discovery_mode = str(payload.get("discovery_mode") or "snapshot").strip().lower()
|
|
|
+ if discovery_mode not in DISCOVERY_MODES:
|
|
|
+ raise ActiveMetadataError("discovery_mode is unsupported")
|
|
|
+ schedule_type, schedule_expression = _schedule(payload)
|
|
|
+ scope = deepcopy(payload.get("scope") or {})
|
|
|
+ if not isinstance(scope, dict):
|
|
|
+ raise ActiveMetadataError("scope must be an object")
|
|
|
+ plan = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "source_uid": _uuid(payload.get("source_uid"), "source_uid"),
|
|
|
+ "name": _text(payload.get("name"), "name"),
|
|
|
+ "source_kind": source_kind,
|
|
|
+ "schedule_type": schedule_type,
|
|
|
+ "schedule_expression": schedule_expression,
|
|
|
+ "discovery_mode": discovery_mode,
|
|
|
+ "scope": scope,
|
|
|
+ "cursor_state": {},
|
|
|
+ "owner_uid": _uuid(payload.get("owner_uid"), "owner_uid"),
|
|
|
+ "enabled": bool(payload.get("enabled", True)),
|
|
|
+ "current_version": 1,
|
|
|
+ "created_by": _uuid(actor_uid, "actor_uid"),
|
|
|
+ "created_at": self.now_factory().isoformat(),
|
|
|
+ }
|
|
|
+ return self.repository.save_plan(plan)
|
|
|
+
|
|
|
+ def list_plans(self):
|
|
|
+ return self.repository.list_plans()
|
|
|
+
|
|
|
+ def execute_source_plans(
|
|
|
+ self,
|
|
|
+ source_uid: str,
|
|
|
+ snapshot: dict[str, Any],
|
|
|
+ *,
|
|
|
+ batch_key: str,
|
|
|
+ actor_uid: str,
|
|
|
+ cursor_after: dict[str, Any] | None = None,
|
|
|
+ ):
|
|
|
+ source_uid = _uuid(source_uid, "source_uid")
|
|
|
+ results = []
|
|
|
+ for plan in self.list_plans():
|
|
|
+ if (
|
|
|
+ plan["source_uid"] != source_uid
|
|
|
+ or plan["source_kind"] != "database"
|
|
|
+ or not plan["enabled"]
|
|
|
+ ):
|
|
|
+ continue
|
|
|
+ results.append(
|
|
|
+ self.execute(
|
|
|
+ plan["uid"],
|
|
|
+ {
|
|
|
+ "snapshot": deepcopy(snapshot),
|
|
|
+ "cursor_after": deepcopy(cursor_after or {}),
|
|
|
+ },
|
|
|
+ batch_key=batch_key,
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ return results
|
|
|
+
|
|
|
+ def get_plan(self, uid: str):
|
|
|
+ result = self.repository.get_plan(_uuid(uid, "plan_uid"))
|
|
|
+ if result is None:
|
|
|
+ raise ActiveMetadataNotFound("active metadata plan was not found")
|
|
|
+ return result
|
|
|
+
|
|
|
+ def list_assets(self, source_uid: str):
|
|
|
+ return self.repository.list_assets(_uuid(source_uid, "source_uid"))
|
|
|
+
|
|
|
+ def list_runs(self, plan_uid: str):
|
|
|
+ self.get_plan(plan_uid)
|
|
|
+ return self.repository.list_runs(plan_uid)
|
|
|
+
|
|
|
+ def list_changes(self, run_uid: str):
|
|
|
+ return self.repository.list_changes(_uuid(run_uid, "run_uid"))
|
|
|
+
|
|
|
+ def list_lineage(self, run_uid: str):
|
|
|
+ return self.repository.list_lineage(_uuid(run_uid, "run_uid"))
|
|
|
+
|
|
|
+ def list_health_signals(self, run_uid: str):
|
|
|
+ return self.repository.list_health_signals(_uuid(run_uid, "run_uid"))
|
|
|
+
|
|
|
+ def list_corrections(self, asset_uid: str | None = None):
|
|
|
+ return self.repository.list_corrections(
|
|
|
+ _uuid(asset_uid, "asset_uid") if asset_uid else None
|
|
|
+ )
|
|
|
+
|
|
|
+ def execute(
|
|
|
+ self,
|
|
|
+ plan_uid: str,
|
|
|
+ payload: Any,
|
|
|
+ *,
|
|
|
+ batch_key: str,
|
|
|
+ actor_uid: str,
|
|
|
+ ):
|
|
|
+ plan = self.get_plan(plan_uid)
|
|
|
+ batch_key = _text(batch_key, "batch_key", 160)
|
|
|
+ existing = self.repository.find_run_by_batch(plan["uid"], batch_key)
|
|
|
+ if existing is not None:
|
|
|
+ return existing
|
|
|
+ if not isinstance(payload, dict):
|
|
|
+ raise ActiveMetadataError("run payload must be an object")
|
|
|
+ _reject_secrets(payload)
|
|
|
+ current_assets = normalize_snapshot(plan["source_uid"], payload.get("snapshot"))
|
|
|
+ previous = {item["asset_key"]: item for item in self.repository.list_assets(plan["source_uid"])}
|
|
|
+ current = {item["key"]: item for item in current_assets}
|
|
|
+ run_uid = self.uid_factory()
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ assets = []
|
|
|
+ versions = []
|
|
|
+ changes = []
|
|
|
+
|
|
|
+ for key, snapshot in current.items():
|
|
|
+ prior = previous.get(key)
|
|
|
+ asset_uid = prior["uid"] if prior else _stable_asset_uid(plan["source_uid"], key)
|
|
|
+ version = int(prior["current_version"]) if prior else 0
|
|
|
+ changed = prior is None or prior["content_hash"] != snapshot["content_hash"]
|
|
|
+ if changed:
|
|
|
+ version += 1
|
|
|
+ versions.append(
|
|
|
+ {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "asset_uid": asset_uid,
|
|
|
+ "version": version,
|
|
|
+ "run_uid": run_uid,
|
|
|
+ "content_hash": snapshot["content_hash"],
|
|
|
+ "snapshot": deepcopy(snapshot),
|
|
|
+ "actor_uid": _uuid(actor_uid, "actor_uid"),
|
|
|
+ "created_at": now,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ asset = {
|
|
|
+ "uid": asset_uid,
|
|
|
+ "source_uid": plan["source_uid"],
|
|
|
+ "asset_key": key,
|
|
|
+ "namespace": snapshot["namespace"],
|
|
|
+ "name": snapshot["name"],
|
|
|
+ "asset_type": snapshot["asset_type"],
|
|
|
+ "lifecycle_status": "active",
|
|
|
+ "current_version": max(version, 1),
|
|
|
+ "content_hash": snapshot["content_hash"],
|
|
|
+ "snapshot": deepcopy(snapshot),
|
|
|
+ "health": deepcopy(prior.get("health", {}) if prior else {}),
|
|
|
+ "last_run_uid": run_uid,
|
|
|
+ "updated_at": now,
|
|
|
+ }
|
|
|
+ assets.append(asset)
|
|
|
+ if prior is None:
|
|
|
+ changes.append(
|
|
|
+ self._change(run_uid, asset, "asset_added", None, None, snapshot)
|
|
|
+ )
|
|
|
+ elif changed:
|
|
|
+ changes.extend(self._field_changes(run_uid, asset, prior["snapshot"], snapshot))
|
|
|
+
|
|
|
+ for key in sorted(set(previous) - set(current)):
|
|
|
+ prior = deepcopy(previous[key])
|
|
|
+ prior["lifecycle_status"] = "deletion_candidate"
|
|
|
+ prior["last_run_uid"] = run_uid
|
|
|
+ prior["updated_at"] = now
|
|
|
+ assets.append(prior)
|
|
|
+ changes.append(
|
|
|
+ self._change(
|
|
|
+ run_uid,
|
|
|
+ prior,
|
|
|
+ "deletion_candidate",
|
|
|
+ None,
|
|
|
+ prior["snapshot"],
|
|
|
+ None,
|
|
|
+ )
|
|
|
+ )
|
|
|
+
|
|
|
+ health_signals = self._health_signals(
|
|
|
+ payload.get("health_signals") or [],
|
|
|
+ assets,
|
|
|
+ run_uid,
|
|
|
+ now,
|
|
|
+ )
|
|
|
+ lineage = self._lineage(payload.get("lineage_sql") or [], run_uid, now)
|
|
|
+ counts = Counter(item["change_type"] for item in changes)
|
|
|
+ run = {
|
|
|
+ "uid": run_uid,
|
|
|
+ "plan_uid": plan["uid"],
|
|
|
+ "batch_key": batch_key,
|
|
|
+ "status": "completed",
|
|
|
+ "attempt_count": 1,
|
|
|
+ "cursor_before": deepcopy(plan.get("cursor_state") or {}),
|
|
|
+ "cursor_after": deepcopy(payload.get("cursor_after") or {}),
|
|
|
+ "snapshot_hash": _hash(current_assets),
|
|
|
+ "statistics": dict(counts),
|
|
|
+ "failure_code": None,
|
|
|
+ "failure_reason": None,
|
|
|
+ "actor_uid": _uuid(actor_uid, "actor_uid"),
|
|
|
+ "started_at": now,
|
|
|
+ "finished_at": now,
|
|
|
+ }
|
|
|
+ return self.repository.apply_discovery(
|
|
|
+ {
|
|
|
+ "run": run,
|
|
|
+ "assets": assets,
|
|
|
+ "versions": versions,
|
|
|
+ "changes": changes,
|
|
|
+ "lineage": lineage,
|
|
|
+ "health_signals": health_signals,
|
|
|
+ }
|
|
|
+ )
|
|
|
+
|
|
|
+ def _change(self, run_uid, asset, change_type, field_name, before, after):
|
|
|
+ return {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "run_uid": run_uid,
|
|
|
+ "asset_uid": asset["uid"],
|
|
|
+ "asset_key": asset["asset_key"],
|
|
|
+ "field_name": field_name,
|
|
|
+ "change_type": change_type,
|
|
|
+ "before_state": deepcopy(before),
|
|
|
+ "after_state": deepcopy(after),
|
|
|
+ "status": "pending",
|
|
|
+ }
|
|
|
+
|
|
|
+ def _field_changes(self, run_uid, asset, before, after):
|
|
|
+ before_fields = {item["name"]: item for item in before.get("fields", [])}
|
|
|
+ after_fields = {item["name"]: item for item in after.get("fields", [])}
|
|
|
+ changes = []
|
|
|
+ for name in sorted(set(after_fields) - set(before_fields)):
|
|
|
+ changes.append(
|
|
|
+ self._change(run_uid, asset, "field_added", name, None, after_fields[name])
|
|
|
+ )
|
|
|
+ for name in sorted(set(before_fields) - set(after_fields)):
|
|
|
+ changes.append(
|
|
|
+ self._change(
|
|
|
+ run_uid,
|
|
|
+ asset,
|
|
|
+ "field_deletion_candidate",
|
|
|
+ name,
|
|
|
+ before_fields[name],
|
|
|
+ None,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ for name in sorted(set(before_fields) & set(after_fields)):
|
|
|
+ if before_fields[name] != after_fields[name]:
|
|
|
+ changes.append(
|
|
|
+ self._change(
|
|
|
+ run_uid,
|
|
|
+ asset,
|
|
|
+ "field_changed",
|
|
|
+ name,
|
|
|
+ before_fields[name],
|
|
|
+ after_fields[name],
|
|
|
+ )
|
|
|
+ )
|
|
|
+ return changes
|
|
|
+
|
|
|
+ def _health_signals(self, raw_signals, assets, run_uid, now):
|
|
|
+ if not isinstance(raw_signals, list):
|
|
|
+ raise ActiveMetadataError("health_signals must be a list")
|
|
|
+ by_key = {item["asset_key"]: item for item in assets}
|
|
|
+ result = []
|
|
|
+ for raw in raw_signals:
|
|
|
+ if not isinstance(raw, dict):
|
|
|
+ raise ActiveMetadataError("health signal must be an object")
|
|
|
+ asset = by_key.get(str(raw.get("asset_key") or ""))
|
|
|
+ if asset is None:
|
|
|
+ raise ActiveMetadataError("health signal asset_key is unknown")
|
|
|
+ signal_type = str(raw.get("signal_type") or "").lower()
|
|
|
+ status = str(raw.get("status") or "unknown").lower()
|
|
|
+ if signal_type not in HEALTH_SIGNAL_TYPES or status not in HEALTH_STATUSES:
|
|
|
+ raise ActiveMetadataError("health signal type or status is unsupported")
|
|
|
+ signal = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "asset_uid": asset["uid"],
|
|
|
+ "run_uid": run_uid,
|
|
|
+ "signal_type": signal_type,
|
|
|
+ "value": raw.get("value"),
|
|
|
+ "status": status,
|
|
|
+ "evidence": deepcopy(raw.get("evidence") or {}),
|
|
|
+ "observed_at": str(raw.get("observed_at") or now),
|
|
|
+ }
|
|
|
+ asset["health"][signal_type] = {
|
|
|
+ "value": signal["value"],
|
|
|
+ "status": status,
|
|
|
+ "observed_at": signal["observed_at"],
|
|
|
+ }
|
|
|
+ result.append(signal)
|
|
|
+ return result
|
|
|
+
|
|
|
+ def _lineage(self, statements, run_uid, now):
|
|
|
+ if not isinstance(statements, list):
|
|
|
+ raise ActiveMetadataError("lineage_sql must be a list")
|
|
|
+ records = []
|
|
|
+ for item in statements:
|
|
|
+ if isinstance(item, str):
|
|
|
+ sql, dialect = item, None
|
|
|
+ elif isinstance(item, dict):
|
|
|
+ sql, dialect = item.get("sql"), item.get("dialect")
|
|
|
+ else:
|
|
|
+ raise ActiveMetadataError("lineage_sql item must be text or object")
|
|
|
+ parsed = parse_sql_field_lineage(sql, dialect=dialect)
|
|
|
+ if parsed["status"] == "failed":
|
|
|
+ records.append(
|
|
|
+ {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "run_uid": run_uid,
|
|
|
+ "parse_status": "failed",
|
|
|
+ "source_asset": None,
|
|
|
+ "source_field": None,
|
|
|
+ "target_asset": None,
|
|
|
+ "target_field": None,
|
|
|
+ "relation_type": "derived_from",
|
|
|
+ "evidence": {"statement_hash": _hash(str(sql or ""))},
|
|
|
+ "failure_reason": parsed["failure_reason"],
|
|
|
+ "created_at": now,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ for edge in parsed["edges"]:
|
|
|
+ records.append(
|
|
|
+ {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "run_uid": run_uid,
|
|
|
+ "parse_status": "resolved",
|
|
|
+ **edge,
|
|
|
+ "failure_reason": None,
|
|
|
+ "created_at": now,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ return records
|
|
|
+
|
|
|
+ def record_failure(
|
|
|
+ self,
|
|
|
+ plan_uid: str,
|
|
|
+ *,
|
|
|
+ batch_key: str,
|
|
|
+ error_code: str,
|
|
|
+ failure_reason: str,
|
|
|
+ actor_uid: str,
|
|
|
+ ):
|
|
|
+ plan = self.get_plan(plan_uid)
|
|
|
+ existing = self.repository.find_run_by_batch(plan["uid"], batch_key)
|
|
|
+ if existing is not None:
|
|
|
+ return existing
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ run_uid = self.uid_factory()
|
|
|
+ run = {
|
|
|
+ "uid": run_uid,
|
|
|
+ "plan_uid": plan["uid"],
|
|
|
+ "batch_key": _text(batch_key, "batch_key", 160),
|
|
|
+ "status": "failed",
|
|
|
+ "attempt_count": 1,
|
|
|
+ "cursor_before": deepcopy(plan.get("cursor_state") or {}),
|
|
|
+ "cursor_after": deepcopy(plan.get("cursor_state") or {}),
|
|
|
+ "snapshot_hash": None,
|
|
|
+ "statistics": {},
|
|
|
+ "failure_code": _text(error_code, "error_code", 80),
|
|
|
+ "failure_reason": _text(failure_reason, "failure_reason", 500),
|
|
|
+ "actor_uid": _uuid(actor_uid, "actor_uid"),
|
|
|
+ "started_at": now,
|
|
|
+ "finished_at": now,
|
|
|
+ }
|
|
|
+ signals = [
|
|
|
+ {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "asset_uid": asset["uid"],
|
|
|
+ "run_uid": run_uid,
|
|
|
+ "signal_type": "task_failure",
|
|
|
+ "value": 1,
|
|
|
+ "status": "critical",
|
|
|
+ "evidence": {"error_code": run["failure_code"]},
|
|
|
+ "observed_at": now,
|
|
|
+ }
|
|
|
+ for asset in self.repository.list_assets(plan["source_uid"])
|
|
|
+ ]
|
|
|
+ return self.repository.record_failed_run(run, signals)
|
|
|
+
|
|
|
+ def submit_correction(self, asset_uid: str, payload: Any, *, actor_uid: str):
|
|
|
+ asset = self.repository.get_asset(_uuid(asset_uid, "asset_uid"))
|
|
|
+ if asset is None:
|
|
|
+ raise ActiveMetadataNotFound("active metadata asset was not found")
|
|
|
+ if not isinstance(payload, dict):
|
|
|
+ raise ActiveMetadataError("correction payload must be an object")
|
|
|
+ _reject_secrets(payload)
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ correction = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "asset_uid": asset["uid"],
|
|
|
+ "field_name": str(payload.get("field_name") or "").strip() or None,
|
|
|
+ "proposed_value": deepcopy(payload.get("proposed_value") or {}),
|
|
|
+ "reason": _text(payload.get("reason"), "reason", 500),
|
|
|
+ "assignee_uid": _uuid(payload.get("assignee_uid"), "assignee_uid"),
|
|
|
+ "status": "pending",
|
|
|
+ "resolution": {},
|
|
|
+ "current_version": 1,
|
|
|
+ "submitted_by": _uuid(actor_uid, "actor_uid"),
|
|
|
+ "created_at": now,
|
|
|
+ "updated_at": now,
|
|
|
+ }
|
|
|
+ return self.repository.save_correction(
|
|
|
+ correction,
|
|
|
+ self._correction_audit(correction, "correction_submitted", actor_uid, None),
|
|
|
+ )
|
|
|
+
|
|
|
+ def resolve_correction(
|
|
|
+ self,
|
|
|
+ correction_uid: str,
|
|
|
+ *,
|
|
|
+ expected_version: int,
|
|
|
+ decision: str,
|
|
|
+ resolution: dict[str, Any],
|
|
|
+ actor_uid: str,
|
|
|
+ ):
|
|
|
+ correction = self.repository.get_correction(_uuid(correction_uid, "correction_uid"))
|
|
|
+ if correction is None:
|
|
|
+ raise ActiveMetadataNotFound("active metadata correction was not found")
|
|
|
+ if int(correction["current_version"]) != int(expected_version):
|
|
|
+ raise ActiveMetadataConflict("correction version conflict")
|
|
|
+ actor_uid = _uuid(actor_uid, "actor_uid")
|
|
|
+ if actor_uid != correction["assignee_uid"]:
|
|
|
+ raise ActiveMetadataError("only the correction assignee may resolve it")
|
|
|
+ if decision not in {"accept", "reject"}:
|
|
|
+ raise ActiveMetadataError("decision must be accept or reject")
|
|
|
+ before = deepcopy(correction)
|
|
|
+ correction.update(
|
|
|
+ {
|
|
|
+ "status": "resolved" if decision == "accept" else "rejected",
|
|
|
+ "resolution": deepcopy(resolution or {}),
|
|
|
+ "current_version": int(correction["current_version"]) + 1,
|
|
|
+ "updated_at": self.now_factory().isoformat(),
|
|
|
+ "resolved_by": actor_uid,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ return self.repository.resolve_correction(
|
|
|
+ correction,
|
|
|
+ self._correction_audit(
|
|
|
+ correction,
|
|
|
+ "correction_resolved",
|
|
|
+ actor_uid,
|
|
|
+ before,
|
|
|
+ ),
|
|
|
+ )
|
|
|
+
|
|
|
+ def _correction_audit(self, correction, action, actor_uid, before):
|
|
|
+ return {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "correction_uid": correction["uid"],
|
|
|
+ "version": correction["current_version"],
|
|
|
+ "action": action,
|
|
|
+ "before_state": deepcopy(before),
|
|
|
+ "after_state": deepcopy(correction),
|
|
|
+ "actor_uid": _uuid(actor_uid, "actor_uid"),
|
|
|
+ "created_at": self.now_factory().isoformat(),
|
|
|
+ }
|