|
|
@@ -0,0 +1,770 @@
|
|
|
+"""Generic business terms, code sets, metrics and physical-field mappings."""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import hashlib
|
|
|
+import json
|
|
|
+import uuid
|
|
|
+import xml.etree.ElementTree as ET
|
|
|
+from copy import deepcopy
|
|
|
+from datetime import UTC, datetime
|
|
|
+from typing import Any
|
|
|
+
|
|
|
+from app.core.common.identifiers import new_governance_uid
|
|
|
+
|
|
|
+ASSET_KINDS = frozenset({"business_term", "code_set", "metric"})
|
|
|
+ASSET_STATUSES = frozenset(
|
|
|
+ {"draft", "in_review", "approved", "published", "superseded", "retired"}
|
|
|
+)
|
|
|
+SECRET_MARKERS = (
|
|
|
+ "password",
|
|
|
+ "secret",
|
|
|
+ "token",
|
|
|
+ "credential",
|
|
|
+ "api_key",
|
|
|
+ "private_key",
|
|
|
+ "authorization",
|
|
|
+ "connection_string",
|
|
|
+)
|
|
|
+ANALYSIS_EXECUTION_FIELDS = frozenset(
|
|
|
+ {"execution_sql", "sql", "query", "script", "executable", "runtime"}
|
|
|
+)
|
|
|
+RDF_NS = "http://www.w3.org/1999/02/22-rdf-syntax-ns#"
|
|
|
+DATAOPS_NS = "https://dataops.local/semantic#"
|
|
|
+
|
|
|
+
|
|
|
+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 _uuid(value: Any, field: str) -> str:
|
|
|
+ try:
|
|
|
+ return str(uuid.UUID(str(value)))
|
|
|
+ except (TypeError, ValueError, AttributeError) as exc:
|
|
|
+ raise ValueError(f"{field} must be a UUID") from exc
|
|
|
+
|
|
|
+
|
|
|
+def _text(value: Any, field: str, maximum: int = 500) -> str:
|
|
|
+ result = str(value or "").strip()
|
|
|
+ if not result or len(result) > maximum:
|
|
|
+ raise ValueError(f"{field} must be between 1 and {maximum} characters")
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+def _optional_text(value: Any, field: str, maximum: int = 500) -> str | None:
|
|
|
+ if value is None:
|
|
|
+ return None
|
|
|
+ return _text(value, field, maximum)
|
|
|
+
|
|
|
+
|
|
|
+def _reject_secrets(value: Any, path: str = "payload") -> None:
|
|
|
+ if isinstance(value, dict):
|
|
|
+ for key, nested in value.items():
|
|
|
+ normalized = str(key).casefold().replace("-", "_")
|
|
|
+ if any(marker in normalized for marker in SECRET_MARKERS):
|
|
|
+ raise ValueError(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 _uuid_list(value: Any, field: str, maximum: int = 500) -> list[str]:
|
|
|
+ if value is None:
|
|
|
+ return []
|
|
|
+ if not isinstance(value, list) or len(value) > maximum:
|
|
|
+ raise ValueError(f"{field} must be a bounded list")
|
|
|
+ result = [_uuid(item, field) for item in value]
|
|
|
+ if len(set(result)) != len(result):
|
|
|
+ raise ValueError(f"{field} contains duplicates")
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+def _aliases(value: Any) -> list[str]:
|
|
|
+ if value is None:
|
|
|
+ return []
|
|
|
+ if not isinstance(value, list) or len(value) > 100:
|
|
|
+ raise ValueError("aliases must be a bounded list")
|
|
|
+ result = [_text(item, "alias", 200) for item in value]
|
|
|
+ if len(set(result)) != len(result):
|
|
|
+ raise ValueError("aliases contains duplicates")
|
|
|
+ return sorted(result)
|
|
|
+
|
|
|
+
|
|
|
+def _code_references(value: Any) -> list[dict[str, str]]:
|
|
|
+ if value is None:
|
|
|
+ return []
|
|
|
+ if not isinstance(value, list) or len(value) > 500:
|
|
|
+ raise ValueError("code_references must be a bounded list")
|
|
|
+ result = []
|
|
|
+ seen = set()
|
|
|
+ for item in value:
|
|
|
+ if not isinstance(item, dict):
|
|
|
+ raise ValueError("code reference must be an object")
|
|
|
+ reference = {
|
|
|
+ "code_set_uid": _uuid(item.get("code_set_uid"), "code_set_uid"),
|
|
|
+ "code": _text(item.get("code"), "code reference", 120),
|
|
|
+ }
|
|
|
+ key = (reference["code_set_uid"], reference["code"])
|
|
|
+ if key in seen:
|
|
|
+ raise ValueError("code_references contains duplicates")
|
|
|
+ seen.add(key)
|
|
|
+ result.append(reference)
|
|
|
+ return sorted(result, key=lambda item: (item["code_set_uid"], item["code"]))
|
|
|
+
|
|
|
+
|
|
|
+def _code_values(value: Any) -> list[dict[str, Any]]:
|
|
|
+ if not isinstance(value, list) or not value or len(value) > 10_000:
|
|
|
+ raise ValueError("values must contain between 1 and 10000 items")
|
|
|
+ result = []
|
|
|
+ codes = set()
|
|
|
+ for item in value:
|
|
|
+ if not isinstance(item, dict):
|
|
|
+ raise ValueError("code value must be an object")
|
|
|
+ code = _text(item.get("code"), "code value", 120)
|
|
|
+ if code in codes:
|
|
|
+ raise ValueError(f"duplicate code value {code}")
|
|
|
+ codes.add(code)
|
|
|
+ result.append(
|
|
|
+ {
|
|
|
+ "code": code,
|
|
|
+ "name": _text(item.get("name"), "code value name", 300),
|
|
|
+ "definition": _optional_text(
|
|
|
+ item.get("definition"), "code value definition", 1000
|
|
|
+ ),
|
|
|
+ "parent_code": _optional_text(
|
|
|
+ item.get("parent_code"), "parent_code", 120
|
|
|
+ ),
|
|
|
+ "status": str(item.get("status") or "active").strip().lower(),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ for item in result:
|
|
|
+ if item["status"] not in {"active", "deprecated"}:
|
|
|
+ raise ValueError("code value status is unsupported")
|
|
|
+ if item["parent_code"] and item["parent_code"] not in codes:
|
|
|
+ raise ValueError(
|
|
|
+ f"parent_code {item['parent_code']} does not reference this code set"
|
|
|
+ )
|
|
|
+ if item["parent_code"] == item["code"]:
|
|
|
+ raise ValueError("parent_code cannot reference itself")
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+def _dimensions(value: Any) -> list[dict[str, str | None]]:
|
|
|
+ if not isinstance(value, list) or not value or len(value) > 100:
|
|
|
+ raise ValueError("dimensions must contain between 1 and 100 items")
|
|
|
+ result = []
|
|
|
+ seen = set()
|
|
|
+ for item in value:
|
|
|
+ if not isinstance(item, dict):
|
|
|
+ raise ValueError("dimension must be an object")
|
|
|
+ code = _text(item.get("code"), "dimension code", 120)
|
|
|
+ if code in seen:
|
|
|
+ raise ValueError(f"duplicate dimension {code}")
|
|
|
+ seen.add(code)
|
|
|
+ result.append(
|
|
|
+ {
|
|
|
+ "code": code,
|
|
|
+ "name": _text(item.get("name"), "dimension name", 300),
|
|
|
+ "definition": _optional_text(
|
|
|
+ item.get("definition"), "dimension definition", 1000
|
|
|
+ ),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+def _normalize_definition(asset_kind: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
|
+ common = {
|
|
|
+ "definition": _text(payload.get("definition"), "definition", 4000),
|
|
|
+ "related_asset_uids": _uuid_list(
|
|
|
+ payload.get("related_asset_uids"), "related_asset_uids"
|
|
|
+ ),
|
|
|
+ "related_data_element_uids": _uuid_list(
|
|
|
+ payload.get("related_data_element_uids"),
|
|
|
+ "related_data_element_uids",
|
|
|
+ ),
|
|
|
+ "standard_version_uids": _uuid_list(
|
|
|
+ payload.get("standard_version_uids"),
|
|
|
+ "standard_version_uids",
|
|
|
+ ),
|
|
|
+ "code_references": _code_references(payload.get("code_references")),
|
|
|
+ }
|
|
|
+ if asset_kind == "business_term":
|
|
|
+ common["aliases"] = _aliases(payload.get("aliases"))
|
|
|
+ elif asset_kind == "code_set":
|
|
|
+ common["values"] = _code_values(payload.get("values"))
|
|
|
+ elif asset_kind == "metric":
|
|
|
+ forbidden = sorted(set(payload) & ANALYSIS_EXECUTION_FIELDS)
|
|
|
+ if forbidden:
|
|
|
+ raise ValueError(
|
|
|
+ "metric is a definition only; analysis execution fields are forbidden"
|
|
|
+ )
|
|
|
+ common.update(
|
|
|
+ {
|
|
|
+ "formula": _text(payload.get("formula"), "formula", 4000),
|
|
|
+ "unit": _text(payload.get("unit"), "unit", 80),
|
|
|
+ "dimensions": _dimensions(payload.get("dimensions")),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ return common
|
|
|
+
|
|
|
+
|
|
|
+def _links(asset_uid: str, version: int, definition: dict[str, Any]) -> list[dict]:
|
|
|
+ result = []
|
|
|
+ for link_type, key, target_type in (
|
|
|
+ ("asset", "related_asset_uids", "active_metadata_asset"),
|
|
|
+ ("data_element", "related_data_element_uids", "data_element"),
|
|
|
+ ("standard", "standard_version_uids", "data_standard_version"),
|
|
|
+ ):
|
|
|
+ result.extend(
|
|
|
+ {
|
|
|
+ "asset_uid": asset_uid,
|
|
|
+ "version": version,
|
|
|
+ "link_type": link_type,
|
|
|
+ "target_type": target_type,
|
|
|
+ "target_uid": uid,
|
|
|
+ "target_key": None,
|
|
|
+ "metadata": {},
|
|
|
+ }
|
|
|
+ for uid in definition.get(key, [])
|
|
|
+ )
|
|
|
+ result.extend(
|
|
|
+ {
|
|
|
+ "asset_uid": asset_uid,
|
|
|
+ "version": version,
|
|
|
+ "link_type": "code_value",
|
|
|
+ "target_type": "code_set",
|
|
|
+ "target_uid": item["code_set_uid"],
|
|
|
+ "target_key": item["code"],
|
|
|
+ "metadata": {},
|
|
|
+ }
|
|
|
+ for item in definition.get("code_references", [])
|
|
|
+ )
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+class SemanticGovernanceService:
|
|
|
+ def __init__(
|
|
|
+ self,
|
|
|
+ repository,
|
|
|
+ *,
|
|
|
+ uid_factory=new_governance_uid,
|
|
|
+ now_factory=lambda: datetime.now(UTC),
|
|
|
+ outbox_enqueue=lambda **_event: None,
|
|
|
+ ):
|
|
|
+ self.repository = repository
|
|
|
+ self.uid_factory = uid_factory
|
|
|
+ self.now_factory = now_factory
|
|
|
+ self.outbox_enqueue = outbox_enqueue
|
|
|
+
|
|
|
+ def _now(self) -> str:
|
|
|
+ return self.now_factory().isoformat()
|
|
|
+
|
|
|
+ def _audit(
|
|
|
+ self,
|
|
|
+ uid: str,
|
|
|
+ version: int,
|
|
|
+ action: str,
|
|
|
+ actor_uid: str,
|
|
|
+ detail: dict[str, Any] | None = None,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ return {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "target_type": "semantic_asset",
|
|
|
+ "target_uid": uid,
|
|
|
+ "target_version": version,
|
|
|
+ "action": action,
|
|
|
+ "actor_uid": _uuid(actor_uid, "actor_uid"),
|
|
|
+ "detail": deepcopy(detail or {}),
|
|
|
+ "created_at": self._now(),
|
|
|
+ }
|
|
|
+
|
|
|
+ def _validate_references(self, definition: dict[str, Any]) -> None:
|
|
|
+ asset_uids = definition.get("related_asset_uids", [])
|
|
|
+ if asset_uids and not self.repository.active_assets_exist(asset_uids):
|
|
|
+ raise ValueError("related asset must exist in active metadata")
|
|
|
+ element_uids = definition.get("related_data_element_uids", [])
|
|
|
+ if element_uids and not self.repository.published_data_elements_exist(
|
|
|
+ element_uids
|
|
|
+ ):
|
|
|
+ raise ValueError("related data element must be published")
|
|
|
+ standard_uids = definition.get("standard_version_uids", [])
|
|
|
+ if standard_uids and not self.repository.published_standard_versions_exist(
|
|
|
+ standard_uids
|
|
|
+ ):
|
|
|
+ raise ValueError("related standard version must be published")
|
|
|
+ for reference in definition.get("code_references", []):
|
|
|
+ if not self.repository.published_code_exists(
|
|
|
+ reference["code_set_uid"], reference["code"]
|
|
|
+ ):
|
|
|
+ raise ValueError("published code reference was not found")
|
|
|
+
|
|
|
+ def create_draft(self, asset_kind: str, payload: Any, *, actor_uid: str):
|
|
|
+ if asset_kind not in ASSET_KINDS:
|
|
|
+ raise ValueError("semantic asset kind is unsupported")
|
|
|
+ if not isinstance(payload, dict):
|
|
|
+ raise ValueError("semantic asset payload must be an object")
|
|
|
+ _reject_secrets(payload)
|
|
|
+ code = _text(payload.get("code"), "code", 120)
|
|
|
+ if self.repository.get_by_code(asset_kind, code) is not None:
|
|
|
+ raise ValueError("semantic asset code already exists")
|
|
|
+ definition = _normalize_definition(asset_kind, payload)
|
|
|
+ self._validate_references(definition)
|
|
|
+ uid = self.uid_factory()
|
|
|
+ now = self._now()
|
|
|
+ asset = {
|
|
|
+ "uid": uid,
|
|
|
+ "asset_kind": asset_kind,
|
|
|
+ "code": code,
|
|
|
+ "name": _text(payload.get("name"), "name", 300),
|
|
|
+ "owner_uid": _uuid(payload.get("owner_uid"), "owner_uid"),
|
|
|
+ "business_domain_uid": _uuid(
|
|
|
+ payload.get("business_domain_uid"), "business_domain_uid"
|
|
|
+ ),
|
|
|
+ "status": "draft",
|
|
|
+ "current_version": 1,
|
|
|
+ "created_by": _uuid(actor_uid, "actor_uid"),
|
|
|
+ "created_at": now,
|
|
|
+ "updated_at": now,
|
|
|
+ }
|
|
|
+ version = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "asset_uid": uid,
|
|
|
+ "version": 1,
|
|
|
+ "status": "draft",
|
|
|
+ "content_hash": _hash(definition),
|
|
|
+ "definition": definition,
|
|
|
+ "change_reason": "initial draft",
|
|
|
+ "created_by": asset["created_by"],
|
|
|
+ "rollback_from_version": None,
|
|
|
+ "created_at": now,
|
|
|
+ "published_at": None,
|
|
|
+ }
|
|
|
+ return self.repository.save_new(
|
|
|
+ asset,
|
|
|
+ version,
|
|
|
+ self._audit(uid, 1, "created", actor_uid),
|
|
|
+ _links(uid, 1, definition),
|
|
|
+ )
|
|
|
+
|
|
|
+ def get(self, uid: str):
|
|
|
+ result = self.repository.get(_uuid(uid, "asset_uid"))
|
|
|
+ if result is None:
|
|
|
+ raise LookupError("semantic asset was not found")
|
|
|
+ return result
|
|
|
+
|
|
|
+ def list(self, *, asset_kind: str | None = None, status: str | None = None):
|
|
|
+ if asset_kind is not None and asset_kind not in ASSET_KINDS:
|
|
|
+ raise ValueError("semantic asset kind is unsupported")
|
|
|
+ if status is not None and status not in ASSET_STATUSES:
|
|
|
+ raise ValueError("semantic asset status is unsupported")
|
|
|
+ return self.repository.list(asset_kind=asset_kind, status=status)
|
|
|
+
|
|
|
+ def list_versions(self, uid: str):
|
|
|
+ self.get(uid)
|
|
|
+ return self.repository.list_versions(uid)
|
|
|
+
|
|
|
+ def get_published_version(self, uid: str, version: int):
|
|
|
+ asset = self.get(uid)
|
|
|
+ record = next(
|
|
|
+ (
|
|
|
+ item
|
|
|
+ for item in self.repository.list_versions(asset["uid"])
|
|
|
+ if int(item["version"]) == int(version)
|
|
|
+ ),
|
|
|
+ None,
|
|
|
+ )
|
|
|
+ if record is None or record["status"] not in {"published", "superseded"}:
|
|
|
+ raise LookupError("published semantic asset version was not found")
|
|
|
+ return {
|
|
|
+ **asset,
|
|
|
+ "status": record["status"],
|
|
|
+ "current_version": int(record["version"]),
|
|
|
+ "definition": deepcopy(record["definition"]),
|
|
|
+ }
|
|
|
+
|
|
|
+ def list_reviews(self, uid: str):
|
|
|
+ self.get(uid)
|
|
|
+ return self.repository.list_reviews(uid)
|
|
|
+
|
|
|
+ def list_audits(self, uid: str):
|
|
|
+ self.get(uid)
|
|
|
+ return self.repository.list_audits(uid)
|
|
|
+
|
|
|
+ def _version(self, asset: dict[str, Any]) -> dict[str, Any]:
|
|
|
+ versions = self.repository.list_versions(asset["uid"])
|
|
|
+ return next(
|
|
|
+ item
|
|
|
+ for item in versions
|
|
|
+ if int(item["version"]) == int(asset["current_version"])
|
|
|
+ )
|
|
|
+
|
|
|
+ def _expected(
|
|
|
+ self, uid: str, expected_version: int
|
|
|
+ ) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
|
+ asset = self.get(uid)
|
|
|
+ if int(expected_version) != int(asset["current_version"]):
|
|
|
+ raise ValueError(
|
|
|
+ f"expected version {expected_version}, "
|
|
|
+ f"current version {asset['current_version']}"
|
|
|
+ )
|
|
|
+ return asset, self._version(asset)
|
|
|
+
|
|
|
+ def submit(
|
|
|
+ self,
|
|
|
+ uid: str,
|
|
|
+ *,
|
|
|
+ expected_version: int,
|
|
|
+ actor_uid: str,
|
|
|
+ ):
|
|
|
+ asset, version = self._expected(uid, expected_version)
|
|
|
+ if asset["status"] != "draft":
|
|
|
+ raise ValueError("only a draft semantic asset can be submitted")
|
|
|
+ asset.update({"status": "in_review", "updated_at": self._now()})
|
|
|
+ version["status"] = "in_review"
|
|
|
+ return self.repository.transition(
|
|
|
+ asset,
|
|
|
+ version,
|
|
|
+ self._audit(uid, asset["current_version"], "submitted", actor_uid),
|
|
|
+ )
|
|
|
+
|
|
|
+ def review(
|
|
|
+ self,
|
|
|
+ uid: str,
|
|
|
+ *,
|
|
|
+ expected_version: int,
|
|
|
+ decision: str,
|
|
|
+ reason: str,
|
|
|
+ actor_uid: str,
|
|
|
+ ):
|
|
|
+ asset, version = self._expected(uid, expected_version)
|
|
|
+ reviewer_uid = _uuid(actor_uid, "actor_uid")
|
|
|
+ if reviewer_uid == str(version["created_by"]):
|
|
|
+ raise ValueError("semantic asset self-review is not allowed")
|
|
|
+ if asset["status"] != "in_review":
|
|
|
+ raise ValueError("only an in-review semantic asset can be reviewed")
|
|
|
+ decision = str(decision or "").strip().lower()
|
|
|
+ if decision not in {"approve", "reject"}:
|
|
|
+ raise ValueError("semantic review decision is unsupported")
|
|
|
+ reason = _text(reason, "review reason", 1000)
|
|
|
+ target_status = "approved" if decision == "approve" else "draft"
|
|
|
+ asset.update({"status": target_status, "updated_at": self._now()})
|
|
|
+ version["status"] = target_status
|
|
|
+ review = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "asset_uid": asset["uid"],
|
|
|
+ "version": asset["current_version"],
|
|
|
+ "decision": decision,
|
|
|
+ "reason": reason,
|
|
|
+ "reviewer_uid": reviewer_uid,
|
|
|
+ "created_at": self._now(),
|
|
|
+ }
|
|
|
+ return self.repository.transition(
|
|
|
+ asset,
|
|
|
+ version,
|
|
|
+ self._audit(
|
|
|
+ uid,
|
|
|
+ asset["current_version"],
|
|
|
+ "approved" if decision == "approve" else "rejected",
|
|
|
+ actor_uid,
|
|
|
+ {"reason": reason},
|
|
|
+ ),
|
|
|
+ review,
|
|
|
+ )
|
|
|
+
|
|
|
+ def publish(
|
|
|
+ self,
|
|
|
+ uid: str,
|
|
|
+ *,
|
|
|
+ expected_version: int,
|
|
|
+ actor_uid: str,
|
|
|
+ ):
|
|
|
+ asset, version = self._expected(uid, expected_version)
|
|
|
+ if asset["status"] != "approved":
|
|
|
+ raise ValueError("only an approved semantic asset can be published")
|
|
|
+ self._validate_references(version["definition"])
|
|
|
+ now = self._now()
|
|
|
+ asset.update({"status": "published", "updated_at": now})
|
|
|
+ version.update({"status": "published", "published_at": now})
|
|
|
+ published = self.repository.transition(
|
|
|
+ asset,
|
|
|
+ version,
|
|
|
+ self._audit(uid, asset["current_version"], "published", actor_uid),
|
|
|
+ )
|
|
|
+ self._enqueue_publication(published)
|
|
|
+ return published
|
|
|
+
|
|
|
+ def revise(
|
|
|
+ self,
|
|
|
+ uid: str,
|
|
|
+ payload: Any,
|
|
|
+ *,
|
|
|
+ expected_version: int,
|
|
|
+ reason: str,
|
|
|
+ actor_uid: str,
|
|
|
+ ):
|
|
|
+ asset, _current = self._expected(uid, expected_version)
|
|
|
+ if asset["status"] not in {"published", "draft"}:
|
|
|
+ raise ValueError("semantic asset cannot be revised in its current status")
|
|
|
+ if not isinstance(payload, dict):
|
|
|
+ raise ValueError("semantic revision payload must be an object")
|
|
|
+ _reject_secrets(payload)
|
|
|
+ merged = {
|
|
|
+ "code": asset["code"],
|
|
|
+ "name": payload.get("name") or asset["name"],
|
|
|
+ "owner_uid": payload.get("owner_uid") or asset["owner_uid"],
|
|
|
+ "business_domain_uid": (
|
|
|
+ payload.get("business_domain_uid") or asset["business_domain_uid"]
|
|
|
+ ),
|
|
|
+ **payload,
|
|
|
+ }
|
|
|
+ definition = _normalize_definition(asset["asset_kind"], merged)
|
|
|
+ self._validate_references(definition)
|
|
|
+ now = self._now()
|
|
|
+ version_number = int(asset["current_version"]) + 1
|
|
|
+ asset.update(
|
|
|
+ {
|
|
|
+ "name": _text(merged.get("name"), "name", 300),
|
|
|
+ "owner_uid": _uuid(merged.get("owner_uid"), "owner_uid"),
|
|
|
+ "business_domain_uid": _uuid(
|
|
|
+ merged.get("business_domain_uid"), "business_domain_uid"
|
|
|
+ ),
|
|
|
+ "status": "draft",
|
|
|
+ "current_version": version_number,
|
|
|
+ "updated_at": now,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ version = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "asset_uid": asset["uid"],
|
|
|
+ "version": version_number,
|
|
|
+ "status": "draft",
|
|
|
+ "content_hash": _hash(definition),
|
|
|
+ "definition": definition,
|
|
|
+ "change_reason": _text(reason, "change reason", 1000),
|
|
|
+ "created_by": _uuid(actor_uid, "actor_uid"),
|
|
|
+ "rollback_from_version": None,
|
|
|
+ "created_at": now,
|
|
|
+ "published_at": None,
|
|
|
+ }
|
|
|
+ return self.repository.append_version(
|
|
|
+ asset,
|
|
|
+ version,
|
|
|
+ self._audit(
|
|
|
+ uid,
|
|
|
+ version_number,
|
|
|
+ "revised",
|
|
|
+ actor_uid,
|
|
|
+ {"reason": version["change_reason"]},
|
|
|
+ ),
|
|
|
+ _links(uid, version_number, definition),
|
|
|
+ )
|
|
|
+
|
|
|
+ def rollback(
|
|
|
+ self,
|
|
|
+ uid: str,
|
|
|
+ *,
|
|
|
+ target_version: int,
|
|
|
+ expected_version: int,
|
|
|
+ reason: str,
|
|
|
+ actor_uid: str,
|
|
|
+ ):
|
|
|
+ asset, _current = self._expected(uid, expected_version)
|
|
|
+ target = next(
|
|
|
+ (
|
|
|
+ item
|
|
|
+ for item in self.repository.list_versions(uid)
|
|
|
+ if int(item["version"]) == int(target_version)
|
|
|
+ ),
|
|
|
+ None,
|
|
|
+ )
|
|
|
+ if target is None or target["status"] not in {"published", "superseded"}:
|
|
|
+ raise ValueError("rollback target must be a published version")
|
|
|
+ now = self._now()
|
|
|
+ version_number = int(asset["current_version"]) + 1
|
|
|
+ definition = deepcopy(target["definition"])
|
|
|
+ asset.update(
|
|
|
+ {
|
|
|
+ "status": "published",
|
|
|
+ "current_version": version_number,
|
|
|
+ "updated_at": now,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ version = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "asset_uid": asset["uid"],
|
|
|
+ "version": version_number,
|
|
|
+ "status": "published",
|
|
|
+ "content_hash": _hash(definition),
|
|
|
+ "definition": definition,
|
|
|
+ "change_reason": _text(reason, "rollback reason", 1000),
|
|
|
+ "created_by": _uuid(actor_uid, "actor_uid"),
|
|
|
+ "rollback_from_version": int(target_version),
|
|
|
+ "created_at": now,
|
|
|
+ "published_at": now,
|
|
|
+ }
|
|
|
+ rolled_back = self.repository.append_version(
|
|
|
+ asset,
|
|
|
+ version,
|
|
|
+ self._audit(
|
|
|
+ uid,
|
|
|
+ version_number,
|
|
|
+ "rolled_back",
|
|
|
+ actor_uid,
|
|
|
+ {
|
|
|
+ "target_version": int(target_version),
|
|
|
+ "reason": version["change_reason"],
|
|
|
+ },
|
|
|
+ ),
|
|
|
+ _links(uid, version_number, definition),
|
|
|
+ )
|
|
|
+ self._enqueue_publication(rolled_back)
|
|
|
+ return rolled_back
|
|
|
+
|
|
|
+ def _enqueue_publication(self, asset: dict[str, Any]) -> None:
|
|
|
+ self.outbox_enqueue(
|
|
|
+ aggregate_type="semantic_asset",
|
|
|
+ aggregate_id=asset["uid"],
|
|
|
+ event_type="semantic_asset.version_published",
|
|
|
+ payload={
|
|
|
+ "uid": asset["uid"],
|
|
|
+ "asset_kind": asset["asset_kind"],
|
|
|
+ "version": asset["current_version"],
|
|
|
+ },
|
|
|
+ )
|
|
|
+
|
|
|
+ def map_physical_field(self, payload: Any, *, actor_uid: str):
|
|
|
+ if not isinstance(payload, dict):
|
|
|
+ raise ValueError("field mapping payload must be an object")
|
|
|
+ _reject_secrets(payload)
|
|
|
+ asset_uid = _uuid(payload.get("asset_uid"), "asset_uid")
|
|
|
+ field_name = _text(payload.get("field_name"), "field_name", 200)
|
|
|
+ data_element_uid = _uuid(
|
|
|
+ payload.get("data_element_uid"), "data_element_uid"
|
|
|
+ )
|
|
|
+ source_asset = self.repository.get_active_metadata_asset(asset_uid)
|
|
|
+ if source_asset is None:
|
|
|
+ raise ValueError("active metadata asset was not found")
|
|
|
+ fields = {
|
|
|
+ str(item.get("name"))
|
|
|
+ for item in source_asset.get("snapshot", {}).get("fields", [])
|
|
|
+ }
|
|
|
+ if field_name not in fields:
|
|
|
+ raise ValueError("physical field was not found in active metadata")
|
|
|
+ element = self.repository.get_data_element(data_element_uid)
|
|
|
+ if element is None or element.get("status") != "published":
|
|
|
+ raise ValueError("physical mapping requires a published data element")
|
|
|
+ now = self._now()
|
|
|
+ mapping = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "asset_uid": asset_uid,
|
|
|
+ "field_name": field_name,
|
|
|
+ "data_element_uid": data_element_uid,
|
|
|
+ "data_element_version": int(element["current_version"]),
|
|
|
+ "owner_uid": _uuid(payload.get("owner_uid"), "owner_uid"),
|
|
|
+ "status": "published",
|
|
|
+ "evidence": deepcopy(payload.get("evidence") or {}),
|
|
|
+ "created_by": _uuid(actor_uid, "actor_uid"),
|
|
|
+ "created_at": now,
|
|
|
+ "updated_at": now,
|
|
|
+ }
|
|
|
+ audit = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "target_type": "field_mapping",
|
|
|
+ "target_uid": mapping["uid"],
|
|
|
+ "target_version": 1,
|
|
|
+ "action": "mapped",
|
|
|
+ "actor_uid": mapping["created_by"],
|
|
|
+ "detail": {
|
|
|
+ "asset_uid": asset_uid,
|
|
|
+ "field_name": field_name,
|
|
|
+ "data_element_uid": data_element_uid,
|
|
|
+ "data_element_version": mapping["data_element_version"],
|
|
|
+ },
|
|
|
+ "created_at": now,
|
|
|
+ }
|
|
|
+ return self.repository.save_field_mapping(mapping, audit)
|
|
|
+
|
|
|
+ def list_field_mappings(
|
|
|
+ self,
|
|
|
+ *,
|
|
|
+ asset_uid: str | None = None,
|
|
|
+ data_element_uid: str | None = None,
|
|
|
+ ):
|
|
|
+ return self.repository.list_field_mappings(
|
|
|
+ asset_uid=_uuid(asset_uid, "asset_uid") if asset_uid else None,
|
|
|
+ data_element_uid=(
|
|
|
+ _uuid(data_element_uid, "data_element_uid")
|
|
|
+ if data_element_uid
|
|
|
+ else None
|
|
|
+ ),
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _safe(value: Any, key: str = "") -> Any:
|
|
|
+ normalized = key.casefold().replace("-", "_")
|
|
|
+ if any(marker in normalized for marker in SECRET_MARKERS):
|
|
|
+ return "[redacted]"
|
|
|
+ if isinstance(value, dict):
|
|
|
+ return {
|
|
|
+ str(item_key): _safe(item_value, str(item_key))
|
|
|
+ for item_key, item_value in sorted(value.items())
|
|
|
+ }
|
|
|
+ if isinstance(value, list):
|
|
|
+ return [_safe(item) for item in value]
|
|
|
+ return value
|
|
|
+
|
|
|
+
|
|
|
+class SemanticExchange:
|
|
|
+ """Deterministic JSON and controlled RDF/XML export for published semantics."""
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _published(assets: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
|
+ return [
|
|
|
+ _safe(item)
|
|
|
+ for item in assets
|
|
|
+ if item.get("status") == "published"
|
|
|
+ ]
|
|
|
+
|
|
|
+ def export_json(self, assets: list[dict[str, Any]]) -> bytes:
|
|
|
+ payload = {
|
|
|
+ "format": "dataops-semantic-governance-json-v1",
|
|
|
+ "assets": self._published(assets),
|
|
|
+ }
|
|
|
+ return _canonical(payload).encode("utf-8")
|
|
|
+
|
|
|
+ def export_rdf(self, assets: list[dict[str, Any]]) -> bytes:
|
|
|
+ ET.register_namespace("rdf", RDF_NS)
|
|
|
+ ET.register_namespace("dataops", DATAOPS_NS)
|
|
|
+ root = ET.Element(f"{{{RDF_NS}}}RDF")
|
|
|
+ tags = {
|
|
|
+ "business_term": "BusinessTerm",
|
|
|
+ "code_set": "CodeSet",
|
|
|
+ "metric": "MetricDefinition",
|
|
|
+ }
|
|
|
+ for asset in self._published(assets):
|
|
|
+ node = ET.SubElement(root, f"{{{DATAOPS_NS}}}{tags[asset['asset_kind']]}")
|
|
|
+ node.set(f"{{{RDF_NS}}}about", f"urn:dataops:semantic:{asset['uid']}")
|
|
|
+ for tag, value in (
|
|
|
+ ("code", asset["code"]),
|
|
|
+ ("name", asset["name"]),
|
|
|
+ ("version", asset["current_version"]),
|
|
|
+ ("definition", asset["definition"]),
|
|
|
+ ):
|
|
|
+ child = ET.SubElement(node, f"{{{DATAOPS_NS}}}{tag}")
|
|
|
+ child.text = (
|
|
|
+ _canonical(value)
|
|
|
+ if isinstance(value, (dict, list))
|
|
|
+ else str(value)
|
|
|
+ )
|
|
|
+ return ET.tostring(root, encoding="utf-8", xml_declaration=True)
|