|
|
@@ -0,0 +1,925 @@
|
|
|
+"""Governed data-product lifecycle over the legacy product and order catalog."""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import copy
|
|
|
+import hashlib
|
|
|
+import json
|
|
|
+import re
|
|
|
+import uuid
|
|
|
+from collections.abc import Callable
|
|
|
+from datetime import datetime
|
|
|
+from typing import Any
|
|
|
+
|
|
|
+from app.core.common.identifiers import new_governance_uid
|
|
|
+from app.core.common.timezone_utils import now_china
|
|
|
+
|
|
|
+PRODUCT_TYPES = frozenset({"database", "api", "file", "data_product"})
|
|
|
+PRODUCT_STATUSES = frozenset({"draft", "in_review", "active", "suspended", "retired"})
|
|
|
+APPLICATION_STATUSES = frozenset(
|
|
|
+ {"draft", "pending_approval", "approved", "rejected", "fulfilled", "cancelled"}
|
|
|
+)
|
|
|
+CONTRACT_STATUSES = frozenset({"draft", "active", "terminated"})
|
|
|
+COMPATIBILITY_MODES = frozenset({"backward", "full", "none"})
|
|
|
+FIELD_TYPES = frozenset(
|
|
|
+ {"string", "integer", "number", "boolean", "date", "datetime", "object", "array", "binary"}
|
|
|
+)
|
|
|
+FEEDBACK_CATEGORIES = frozenset({"quality", "freshness", "usability", "documentation", "service"})
|
|
|
+FEEDBACK_STATUSES = frozenset({"open", "triaged", "in_progress", "resolved", "closed"})
|
|
|
+CODE_PATTERN = re.compile(r"^[A-Z][A-Z0-9_]{2,119}$")
|
|
|
+
|
|
|
+
|
|
|
+def _closed(value: Any, allowed: set[str], label: str) -> dict[str, Any]:
|
|
|
+ if not isinstance(value, dict):
|
|
|
+ raise ValueError(f"{label} must be an object")
|
|
|
+ unknown = sorted(set(value) - allowed)
|
|
|
+ if unknown:
|
|
|
+ raise ValueError(f"{label} contains unsupported fields: {', '.join(unknown)}")
|
|
|
+ return copy.deepcopy(value)
|
|
|
+
|
|
|
+
|
|
|
+def _string(value: Any, label: str, maximum: int = 1000) -> str:
|
|
|
+ if not isinstance(value, str) or not value.strip():
|
|
|
+ raise ValueError(f"{label} is required")
|
|
|
+ result = value.strip()
|
|
|
+ if len(result) > maximum:
|
|
|
+ raise ValueError(f"{label} exceeds {maximum} characters")
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+def _uid(value: Any, label: str) -> str:
|
|
|
+ try:
|
|
|
+ return str(uuid.UUID(str(value)))
|
|
|
+ except (TypeError, ValueError, AttributeError) as error:
|
|
|
+ raise ValueError(f"{label} must be a UUID") from error
|
|
|
+
|
|
|
+
|
|
|
+def _list(value: Any, label: str, minimum: int = 0) -> list[Any]:
|
|
|
+ if not isinstance(value, list) or len(value) < minimum:
|
|
|
+ raise ValueError(f"{label} must contain at least {minimum} items")
|
|
|
+ return copy.deepcopy(value)
|
|
|
+
|
|
|
+
|
|
|
+def _number(value: Any, label: str, minimum: float, maximum: float) -> float:
|
|
|
+ try:
|
|
|
+ result = float(value)
|
|
|
+ except (TypeError, ValueError) as error:
|
|
|
+ raise ValueError(f"{label} must be numeric") from error
|
|
|
+ if result < minimum or result > maximum:
|
|
|
+ raise ValueError(f"{label} is outside the allowed range")
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+def _canonical_hash(value: Any) -> str:
|
|
|
+ encoded = json.dumps(
|
|
|
+ value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
|
|
|
+ ).encode("utf-8")
|
|
|
+ return hashlib.sha256(encoded).hexdigest()
|
|
|
+
|
|
|
+
|
|
|
+def _normalize_sla(value: Any) -> dict[str, Any]:
|
|
|
+ body = _closed(
|
|
|
+ value,
|
|
|
+ {"availability_target", "freshness_hours", "support_tier"},
|
|
|
+ "product SLA",
|
|
|
+ )
|
|
|
+ support_tier = _string(body.get("support_tier"), "support_tier", 40)
|
|
|
+ if support_tier not in {"best_effort", "business_hours", "24x7"}:
|
|
|
+ raise ValueError("unsupported support tier")
|
|
|
+ return {
|
|
|
+ "availability_target": _number(
|
|
|
+ body.get("availability_target"), "availability_target", 0, 100
|
|
|
+ ),
|
|
|
+ "freshness_hours": _number(
|
|
|
+ body.get("freshness_hours"), "freshness_hours", 0, 8760
|
|
|
+ ),
|
|
|
+ "support_tier": support_tier,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def _normalize_schema(value: Any) -> dict[str, Any]:
|
|
|
+ body = _closed(value, {"fields"}, "contract schema")
|
|
|
+ fields = []
|
|
|
+ names = set()
|
|
|
+ for item in _list(body.get("fields"), "contract fields", minimum=1):
|
|
|
+ field = _closed(item, {"name", "type", "nullable"}, "contract field")
|
|
|
+ name = _string(field.get("name"), "field name", 200)
|
|
|
+ field_type = _string(field.get("type"), "field type", 30)
|
|
|
+ if field_type not in FIELD_TYPES:
|
|
|
+ raise ValueError("unsupported contract field type")
|
|
|
+ if name in names:
|
|
|
+ raise ValueError("contract field names must be unique")
|
|
|
+ if not isinstance(field.get("nullable"), bool):
|
|
|
+ raise ValueError("field nullable must be boolean")
|
|
|
+ names.add(name)
|
|
|
+ fields.append(
|
|
|
+ {"name": name, "type": field_type, "nullable": field["nullable"]}
|
|
|
+ )
|
|
|
+ return {"fields": fields}
|
|
|
+
|
|
|
+
|
|
|
+def _normalize_contract(value: Any) -> dict[str, Any]:
|
|
|
+ body = _closed(
|
|
|
+ value,
|
|
|
+ {
|
|
|
+ "schema",
|
|
|
+ "delivery",
|
|
|
+ "quality_terms",
|
|
|
+ "sla_terms",
|
|
|
+ "usage_terms",
|
|
|
+ "compatibility_mode",
|
|
|
+ "change_reason",
|
|
|
+ },
|
|
|
+ "data contract",
|
|
|
+ )
|
|
|
+ delivery = _closed(body.get("delivery"), {"mode", "format"}, "delivery")
|
|
|
+ mode = _string(delivery.get("mode"), "delivery mode", 40)
|
|
|
+ if mode not in {"table", "api", "file", "product"}:
|
|
|
+ raise ValueError("unsupported delivery mode")
|
|
|
+ quality = _closed(
|
|
|
+ body.get("quality_terms"), {"minimum_score"}, "quality terms"
|
|
|
+ )
|
|
|
+ sla = _closed(
|
|
|
+ body.get("sla_terms"),
|
|
|
+ {"freshness_hours", "availability_target"},
|
|
|
+ "SLA terms",
|
|
|
+ )
|
|
|
+ usage = _closed(
|
|
|
+ body.get("usage_terms"), {"purpose", "retention_days"}, "usage terms"
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ retention_days = int(usage.get("retention_days"))
|
|
|
+ except (TypeError, ValueError) as error:
|
|
|
+ raise ValueError("retention_days must be an integer") from error
|
|
|
+ if retention_days < 1 or retention_days > 36500:
|
|
|
+ raise ValueError("retention_days is outside the allowed range")
|
|
|
+ compatibility_mode = _string(
|
|
|
+ body.get("compatibility_mode"), "compatibility_mode", 20
|
|
|
+ )
|
|
|
+ if compatibility_mode not in COMPATIBILITY_MODES:
|
|
|
+ raise ValueError("unsupported compatibility mode")
|
|
|
+ return {
|
|
|
+ "schema": _normalize_schema(body.get("schema")),
|
|
|
+ "delivery": {
|
|
|
+ "mode": mode,
|
|
|
+ "format": _string(delivery.get("format"), "delivery format", 40),
|
|
|
+ },
|
|
|
+ "quality_terms": {
|
|
|
+ "minimum_score": _number(
|
|
|
+ quality.get("minimum_score"), "minimum_score", 0, 100
|
|
|
+ )
|
|
|
+ },
|
|
|
+ "sla_terms": {
|
|
|
+ "freshness_hours": _number(
|
|
|
+ sla.get("freshness_hours"), "freshness_hours", 0, 8760
|
|
|
+ ),
|
|
|
+ "availability_target": _number(
|
|
|
+ sla.get("availability_target"), "availability_target", 0, 100
|
|
|
+ ),
|
|
|
+ },
|
|
|
+ "usage_terms": {
|
|
|
+ "purpose": _string(usage.get("purpose"), "usage purpose", 500),
|
|
|
+ "retention_days": retention_days,
|
|
|
+ },
|
|
|
+ "compatibility_mode": compatibility_mode,
|
|
|
+ "change_reason": _string(body.get("change_reason"), "change_reason", 1000),
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def _compatibility(previous: dict[str, Any], candidate: dict[str, Any]):
|
|
|
+ mode = candidate["compatibility_mode"]
|
|
|
+ if mode == "none":
|
|
|
+ return {"compatible": True, "breaking_changes": [], "mode": mode}
|
|
|
+ old_fields = {item["name"]: item for item in previous["schema"]["fields"]}
|
|
|
+ new_fields = {item["name"]: item for item in candidate["schema"]["fields"]}
|
|
|
+ changes = []
|
|
|
+ for name, old in old_fields.items():
|
|
|
+ new = new_fields.get(name)
|
|
|
+ if new is None:
|
|
|
+ changes.append(f"field {name} was removed")
|
|
|
+ continue
|
|
|
+ if new["type"] != old["type"]:
|
|
|
+ changes.append(f"field {name} changed type")
|
|
|
+ if old["nullable"] and not new["nullable"]:
|
|
|
+ changes.append(f"field {name} tightened nullability")
|
|
|
+ if mode == "full":
|
|
|
+ for name, field in new_fields.items():
|
|
|
+ if name not in old_fields and not field["nullable"]:
|
|
|
+ changes.append(f"required field {name} was added")
|
|
|
+ return {"compatible": not changes, "breaking_changes": changes, "mode": mode}
|
|
|
+
|
|
|
+
|
|
|
+class ProductGovernanceService:
|
|
|
+ """Coordinate product governance without granting data access."""
|
|
|
+
|
|
|
+ def __init__(
|
|
|
+ self,
|
|
|
+ repository,
|
|
|
+ *,
|
|
|
+ approval_gateway,
|
|
|
+ uid_factory: Callable[[], str] = new_governance_uid,
|
|
|
+ now_factory: Callable[[], datetime] = now_china,
|
|
|
+ commit: Callable[[], None] = lambda: None,
|
|
|
+ rollback: Callable[[], None] = lambda: None,
|
|
|
+ ):
|
|
|
+ self.repository = repository
|
|
|
+ self.approval_gateway = approval_gateway
|
|
|
+ self.uid_factory = uid_factory
|
|
|
+ self.now_factory = now_factory
|
|
|
+ self.commit = commit
|
|
|
+ self.rollback = rollback
|
|
|
+
|
|
|
+ def register_product(self, payload: Any, *, actor_uid: str):
|
|
|
+ body = _closed(
|
|
|
+ payload,
|
|
|
+ {
|
|
|
+ "legacy_product_id",
|
|
|
+ "product_code",
|
|
|
+ "name",
|
|
|
+ "product_type",
|
|
|
+ "owner_uid",
|
|
|
+ "business_domain_uid",
|
|
|
+ "description",
|
|
|
+ "quality_target",
|
|
|
+ "sla",
|
|
|
+ },
|
|
|
+ "governed product",
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ legacy_product_id = int(body.get("legacy_product_id"))
|
|
|
+ except (TypeError, ValueError) as error:
|
|
|
+ raise ValueError("legacy_product_id must be an integer") from error
|
|
|
+ if not self.repository.legacy_product_exists(legacy_product_id):
|
|
|
+ raise LookupError("legacy data product was not found")
|
|
|
+ code = _string(body.get("product_code"), "product_code", 120).upper()
|
|
|
+ if not CODE_PATTERN.fullmatch(code):
|
|
|
+ raise ValueError("product_code is invalid")
|
|
|
+ product_type = _string(body.get("product_type"), "product_type", 30)
|
|
|
+ if product_type not in PRODUCT_TYPES:
|
|
|
+ raise ValueError("unsupported product type")
|
|
|
+ owner = _uid(body.get("owner_uid"), "owner_uid")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ if self.repository.users_available({owner, actor}) != {owner, actor}:
|
|
|
+ raise ValueError("product owner or actor is unavailable")
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ record = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "legacy_product_id": legacy_product_id,
|
|
|
+ "product_code": code,
|
|
|
+ "name": _string(body.get("name"), "name", 300),
|
|
|
+ "product_type": product_type,
|
|
|
+ "owner_uid": owner,
|
|
|
+ "business_domain_uid": _uid(
|
|
|
+ body.get("business_domain_uid"), "business_domain_uid"
|
|
|
+ ),
|
|
|
+ "description": _string(body.get("description"), "description", 2000),
|
|
|
+ "quality_target": _number(
|
|
|
+ body.get("quality_target"), "quality_target", 0, 100
|
|
|
+ ),
|
|
|
+ "sla": _normalize_sla(body.get("sla")),
|
|
|
+ "status": "draft",
|
|
|
+ "current_version": 1,
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now,
|
|
|
+ "updated_by": actor,
|
|
|
+ "updated_at": now,
|
|
|
+ "retired_at": None,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ result = self.repository.create_product(record)
|
|
|
+ self.repository.add_event(
|
|
|
+ record["uid"], "product_registered", actor, {}, version=1
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def list_products(self, **filters):
|
|
|
+ return self.repository.list_products(**filters)
|
|
|
+
|
|
|
+ def product_detail(self, uid: str):
|
|
|
+ result = self.repository.product_detail(_uid(uid, "product_uid"))
|
|
|
+ if result is None:
|
|
|
+ raise LookupError("governed product was not found")
|
|
|
+ return result
|
|
|
+
|
|
|
+ def transition_product(
|
|
|
+ self, uid: str, payload: Any, *, expected_version: int, actor_uid: str
|
|
|
+ ):
|
|
|
+ body = _closed(payload, {"action", "reason"}, "product transition")
|
|
|
+ product_uid = _uid(uid, "product_uid")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ product = self.repository.get_product(product_uid)
|
|
|
+ if product is None:
|
|
|
+ raise LookupError("governed product was not found")
|
|
|
+ if actor != product["owner_uid"]:
|
|
|
+ raise PermissionError("only the product owner can transition lifecycle")
|
|
|
+ action = _string(body.get("action"), "action", 30)
|
|
|
+ transitions = {
|
|
|
+ ("draft", "submit_review"): ("in_review", "product_submitted"),
|
|
|
+ ("in_review", "activate"): ("active", "product_activated"),
|
|
|
+ ("draft", "activate"): ("active", "product_activated"),
|
|
|
+ ("active", "suspend"): ("suspended", "product_suspended"),
|
|
|
+ ("suspended", "reactivate"): ("active", "product_reactivated"),
|
|
|
+ ("draft", "retire"): ("retired", "product_retired"),
|
|
|
+ ("in_review", "retire"): ("retired", "product_retired"),
|
|
|
+ ("active", "retire"): ("retired", "product_retired"),
|
|
|
+ ("suspended", "retire"): ("retired", "product_retired"),
|
|
|
+ }
|
|
|
+ target = transitions.get((product["status"], action))
|
|
|
+ if target is None:
|
|
|
+ raise RuntimeError("product lifecycle transition is not allowed")
|
|
|
+ if action in {"activate", "reactivate"}:
|
|
|
+ contract = self.repository.contract_for_product(product_uid)
|
|
|
+ certificate = self.repository.latest_certificate(product_uid)
|
|
|
+ if not contract or contract["status"] != "active":
|
|
|
+ raise RuntimeError("an active data contract is required")
|
|
|
+ if not certificate or certificate["status"] != "qualified":
|
|
|
+ raise RuntimeError("a qualified product certificate is required")
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ product.update(
|
|
|
+ {
|
|
|
+ "status": target[0],
|
|
|
+ "updated_by": actor,
|
|
|
+ "updated_at": now,
|
|
|
+ "retired_at": now if target[0] == "retired" else None,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ result = self.repository.update_product(
|
|
|
+ product,
|
|
|
+ int(expected_version),
|
|
|
+ target[1],
|
|
|
+ actor,
|
|
|
+ {"reason": _string(body.get("reason"), "reason", 1000)},
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def create_application(self, payload: Any, *, actor_uid: str):
|
|
|
+ body = _closed(
|
|
|
+ payload,
|
|
|
+ {
|
|
|
+ "title",
|
|
|
+ "source_type",
|
|
|
+ "source_ref",
|
|
|
+ "business_domain_uid",
|
|
|
+ "purpose",
|
|
|
+ "requested_fields",
|
|
|
+ },
|
|
|
+ "product application",
|
|
|
+ )
|
|
|
+ source_type = _string(body.get("source_type"), "source_type", 30)
|
|
|
+ if source_type not in PRODUCT_TYPES:
|
|
|
+ raise ValueError("unsupported application source type")
|
|
|
+ source_ref = body.get("source_ref")
|
|
|
+ if not isinstance(source_ref, dict) or not source_ref:
|
|
|
+ raise ValueError("source_ref must be a non-empty object")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ if self.repository.users_available({actor}) != {actor}:
|
|
|
+ raise ValueError("application requester is unavailable")
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ record = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "application_code": f"DPA-{self.now_factory():%Y%m%d}-{self.uid_factory()[-8:].upper()}",
|
|
|
+ "title": _string(body.get("title"), "title", 300),
|
|
|
+ "source_type": source_type,
|
|
|
+ "source_ref": copy.deepcopy(source_ref),
|
|
|
+ "business_domain_uid": _uid(
|
|
|
+ body.get("business_domain_uid"), "business_domain_uid"
|
|
|
+ ),
|
|
|
+ "purpose": _string(body.get("purpose"), "purpose", 1000),
|
|
|
+ "requested_fields": sorted(
|
|
|
+ {
|
|
|
+ _string(item, "requested field", 200)
|
|
|
+ for item in _list(
|
|
|
+ body.get("requested_fields"), "requested_fields", minimum=1
|
|
|
+ )
|
|
|
+ }
|
|
|
+ ),
|
|
|
+ "status": "draft",
|
|
|
+ "approval_task_uid": None,
|
|
|
+ "product_uid": None,
|
|
|
+ "current_version": 1,
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now,
|
|
|
+ "updated_by": actor,
|
|
|
+ "updated_at": now,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ result = self.repository.create_application(record)
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def list_applications(self, **filters):
|
|
|
+ return self.repository.list_applications(**filters)
|
|
|
+
|
|
|
+ def submit_application(
|
|
|
+ self, uid: str, payload: Any, *, expected_version: int, actor_uid: str
|
|
|
+ ):
|
|
|
+ body = _closed(payload, {"workflow_uid"}, "application submission")
|
|
|
+ application_uid = _uid(uid, "application_uid")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ application = self.repository.get_application(application_uid)
|
|
|
+ if application is None:
|
|
|
+ raise LookupError("product application was not found")
|
|
|
+ if application["status"] != "draft" or application["created_by"] != actor:
|
|
|
+ raise PermissionError("only the requester can submit a draft application")
|
|
|
+ task = self.approval_gateway.create_product_task(
|
|
|
+ application,
|
|
|
+ _uid(body.get("workflow_uid"), "workflow_uid"),
|
|
|
+ actor,
|
|
|
+ )
|
|
|
+ application.update(
|
|
|
+ {
|
|
|
+ "status": "pending_approval",
|
|
|
+ "approval_task_uid": task["uid"],
|
|
|
+ "updated_by": actor,
|
|
|
+ "updated_at": self.now_factory().isoformat(),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ result = self.repository.update_application(
|
|
|
+ application,
|
|
|
+ int(expected_version),
|
|
|
+ "application_submitted",
|
|
|
+ actor,
|
|
|
+ {"approval_task_uid": task["uid"]},
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def reconcile_application(
|
|
|
+ self, uid: str, *, expected_version: int, actor_uid: str
|
|
|
+ ):
|
|
|
+ application_uid = _uid(uid, "application_uid")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ application = self.repository.get_application(application_uid)
|
|
|
+ if application is None:
|
|
|
+ raise LookupError("product application was not found")
|
|
|
+ if application["status"] != "pending_approval":
|
|
|
+ raise RuntimeError("application is not waiting for approval")
|
|
|
+ task = self.approval_gateway.get_task(application["approval_task_uid"])
|
|
|
+ if not task or task["status"] not in {"approved", "rejected"}:
|
|
|
+ raise RuntimeError("approval task has no final decision")
|
|
|
+ application.update(
|
|
|
+ {
|
|
|
+ "status": task["status"],
|
|
|
+ "updated_by": actor,
|
|
|
+ "updated_at": self.now_factory().isoformat(),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ result = self.repository.update_application(
|
|
|
+ application,
|
|
|
+ int(expected_version),
|
|
|
+ f"application_{task['status']}",
|
|
|
+ actor,
|
|
|
+ {"approval_task_uid": task["uid"]},
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def fulfill_application(
|
|
|
+ self,
|
|
|
+ uid: str,
|
|
|
+ product_uid: str,
|
|
|
+ *,
|
|
|
+ expected_version: int,
|
|
|
+ actor_uid: str,
|
|
|
+ ):
|
|
|
+ application_uid = _uid(uid, "application_uid")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ application = self.repository.get_application(application_uid)
|
|
|
+ product = self.repository.get_product(_uid(product_uid, "product_uid"))
|
|
|
+ if application is None or product is None:
|
|
|
+ raise LookupError("application or governed product was not found")
|
|
|
+ if application["status"] != "approved":
|
|
|
+ raise RuntimeError("only an approved application can be fulfilled")
|
|
|
+ application.update(
|
|
|
+ {
|
|
|
+ "status": "fulfilled",
|
|
|
+ "product_uid": product["uid"],
|
|
|
+ "updated_by": actor,
|
|
|
+ "updated_at": self.now_factory().isoformat(),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ result = self.repository.update_application(
|
|
|
+ application,
|
|
|
+ int(expected_version),
|
|
|
+ "application_fulfilled",
|
|
|
+ actor,
|
|
|
+ {"product_uid": product["uid"], "grants_data_access": False},
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def create_contract(self, product_uid: str, payload: Any, *, actor_uid: str):
|
|
|
+ uid = _uid(product_uid, "product_uid")
|
|
|
+ product = self.repository.get_product(uid)
|
|
|
+ if product is None:
|
|
|
+ raise LookupError("governed product was not found")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ if actor != product["owner_uid"]:
|
|
|
+ raise PermissionError("only the product owner can author a contract")
|
|
|
+ if self.repository.contract_for_product(uid):
|
|
|
+ raise RuntimeError("the product already has a data contract")
|
|
|
+ definition = _normalize_contract(payload)
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ contract_uid = self.uid_factory()
|
|
|
+ version = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "contract_uid": contract_uid,
|
|
|
+ "version": 1,
|
|
|
+ "status": "draft",
|
|
|
+ "definition": definition,
|
|
|
+ "content_hash": _canonical_hash(definition),
|
|
|
+ "compatibility": {"compatible": True, "breaking_changes": [], "mode": definition["compatibility_mode"]},
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now,
|
|
|
+ "published_by": None,
|
|
|
+ "published_at": None,
|
|
|
+ }
|
|
|
+ contract = {
|
|
|
+ "uid": contract_uid,
|
|
|
+ "product_uid": uid,
|
|
|
+ "contract_code": f"DPC-{product['product_code']}",
|
|
|
+ "status": "draft",
|
|
|
+ "current_version": 1,
|
|
|
+ "active_version_uid": None,
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now,
|
|
|
+ "updated_by": actor,
|
|
|
+ "updated_at": now,
|
|
|
+ "terminated_at": None,
|
|
|
+ "termination_reason": None,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ self.repository.create_contract(contract, version)
|
|
|
+ self.repository.add_event(uid, "contract_created", actor, {"contract_uid": contract_uid})
|
|
|
+ self.commit()
|
|
|
+ return {**contract, "latest_version": version}
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def check_contract_compatibility(self, contract_uid: str, payload: Any):
|
|
|
+ uid = _uid(contract_uid, "contract_uid")
|
|
|
+ contract = self.repository.get_contract(uid)
|
|
|
+ if contract is None:
|
|
|
+ raise LookupError("data contract was not found")
|
|
|
+ candidate = _normalize_contract(payload)
|
|
|
+ active = self.repository.active_contract_version(uid)
|
|
|
+ if active is None:
|
|
|
+ return {"compatible": True, "breaking_changes": [], "mode": candidate["compatibility_mode"]}
|
|
|
+ return _compatibility(active["definition"], candidate)
|
|
|
+
|
|
|
+ def revise_contract(
|
|
|
+ self, contract_uid: str, payload: Any, *, expected_version: int, actor_uid: str
|
|
|
+ ):
|
|
|
+ uid = _uid(contract_uid, "contract_uid")
|
|
|
+ contract = self.repository.get_contract(uid)
|
|
|
+ if contract is None:
|
|
|
+ raise LookupError("data contract was not found")
|
|
|
+ product = self.repository.get_product(contract["product_uid"])
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ if actor != product["owner_uid"] or contract["status"] == "terminated":
|
|
|
+ raise PermissionError("contract cannot be revised by this actor")
|
|
|
+ definition = _normalize_contract(payload)
|
|
|
+ compatibility = self.check_contract_compatibility(uid, payload)
|
|
|
+ next_version = int(expected_version) + 1
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ version = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "contract_uid": uid,
|
|
|
+ "version": next_version,
|
|
|
+ "status": "draft",
|
|
|
+ "definition": definition,
|
|
|
+ "content_hash": _canonical_hash(definition),
|
|
|
+ "compatibility": compatibility,
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now,
|
|
|
+ "published_by": None,
|
|
|
+ "published_at": None,
|
|
|
+ }
|
|
|
+ revised = {
|
|
|
+ **contract,
|
|
|
+ "current_version": next_version,
|
|
|
+ "updated_by": actor,
|
|
|
+ "updated_at": now,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ self.repository.revise_contract(revised, version, int(expected_version))
|
|
|
+ self.repository.add_event(
|
|
|
+ contract["product_uid"],
|
|
|
+ "contract_revised",
|
|
|
+ actor,
|
|
|
+ {"contract_uid": uid, "version": next_version, **compatibility},
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return {**revised, "latest_version": version}
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def publish_contract(
|
|
|
+ self, contract_uid: str, *, expected_version: int, actor_uid: str
|
|
|
+ ):
|
|
|
+ uid = _uid(contract_uid, "contract_uid")
|
|
|
+ contract = self.repository.get_contract(uid)
|
|
|
+ if contract is None:
|
|
|
+ raise LookupError("data contract was not found")
|
|
|
+ product = self.repository.get_product(contract["product_uid"])
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ if actor != product["owner_uid"]:
|
|
|
+ raise PermissionError("only the product owner can publish a contract")
|
|
|
+ if int(contract["current_version"]) != int(expected_version):
|
|
|
+ raise RuntimeError("contract version conflict")
|
|
|
+ version = self.repository.contract_version(uid, int(expected_version))
|
|
|
+ if version is None or version["status"] != "draft":
|
|
|
+ raise RuntimeError("contract version is not publishable")
|
|
|
+ if not version["compatibility"]["compatible"]:
|
|
|
+ raise RuntimeError("breaking contract version cannot be published")
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ version.update(
|
|
|
+ {"status": "published", "published_by": actor, "published_at": now}
|
|
|
+ )
|
|
|
+ published = {
|
|
|
+ **contract,
|
|
|
+ "status": "active",
|
|
|
+ "active_version_uid": version["uid"],
|
|
|
+ "updated_by": actor,
|
|
|
+ "updated_at": now,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ self.repository.publish_contract(published, version, int(expected_version))
|
|
|
+ self.repository.add_event(
|
|
|
+ contract["product_uid"],
|
|
|
+ "contract_published",
|
|
|
+ actor,
|
|
|
+ {"contract_uid": uid, "version": version["version"]},
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return {**published, "active_version": version}
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def terminate_contract(
|
|
|
+ self, contract_uid: str, payload: Any, *, expected_version: int, actor_uid: str
|
|
|
+ ):
|
|
|
+ body = _closed(payload, {"reason", "evidence_refs"}, "contract termination")
|
|
|
+ uid = _uid(contract_uid, "contract_uid")
|
|
|
+ contract = self.repository.get_contract(uid)
|
|
|
+ if contract is None:
|
|
|
+ raise LookupError("data contract was not found")
|
|
|
+ product = self.repository.get_product(contract["product_uid"])
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ if actor != product["owner_uid"] or contract["status"] != "active":
|
|
|
+ raise PermissionError("active contract cannot be terminated by this actor")
|
|
|
+ evidence = [
|
|
|
+ _string(item, "evidence reference", 1000)
|
|
|
+ for item in _list(body.get("evidence_refs"), "evidence_refs", minimum=1)
|
|
|
+ ]
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ terminated = {
|
|
|
+ **contract,
|
|
|
+ "status": "terminated",
|
|
|
+ "updated_by": actor,
|
|
|
+ "updated_at": now,
|
|
|
+ "terminated_at": now,
|
|
|
+ "termination_reason": _string(body.get("reason"), "reason", 1000),
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ self.repository.terminate_contract(terminated, int(expected_version))
|
|
|
+ self.repository.add_event(
|
|
|
+ contract["product_uid"],
|
|
|
+ "contract_terminated",
|
|
|
+ actor,
|
|
|
+ {"reason": terminated["termination_reason"], "evidence_refs": evidence},
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return terminated
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def generate_certificate(self, product_uid: str, payload: Any, *, actor_uid: str):
|
|
|
+ body = _closed(payload, {"approval_task_uid", "evidence"}, "certificate request")
|
|
|
+ uid = _uid(product_uid, "product_uid")
|
|
|
+ product = self.repository.get_product(uid)
|
|
|
+ if product is None:
|
|
|
+ raise LookupError("governed product was not found")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ if actor != product["owner_uid"]:
|
|
|
+ raise PermissionError("only the product owner can issue a certificate")
|
|
|
+ contract = self.repository.contract_for_product(uid)
|
|
|
+ if not contract or contract["status"] != "active":
|
|
|
+ raise RuntimeError("an active data contract is required")
|
|
|
+ approval_uid = _uid(body.get("approval_task_uid"), "approval_task_uid")
|
|
|
+ approval = self.approval_gateway.get_task(approval_uid)
|
|
|
+ if not approval or approval["status"] != "approved":
|
|
|
+ raise RuntimeError("approved work-center evidence is required")
|
|
|
+ references = _closed(
|
|
|
+ body.get("evidence"),
|
|
|
+ {
|
|
|
+ "quality_run_uid",
|
|
|
+ "lineage_uids",
|
|
|
+ "rule_version_uids",
|
|
|
+ "workflow_run_uids",
|
|
|
+ },
|
|
|
+ "certificate evidence",
|
|
|
+ )
|
|
|
+ references = {
|
|
|
+ "quality_run_uid": _uid(references.get("quality_run_uid"), "quality_run_uid"),
|
|
|
+ "lineage_uids": [_uid(item, "lineage_uid") for item in _list(references.get("lineage_uids"), "lineage_uids", 1)],
|
|
|
+ "rule_version_uids": [_uid(item, "rule_version_uid") for item in _list(references.get("rule_version_uids"), "rule_version_uids", 1)],
|
|
|
+ "workflow_run_uids": [_uid(item, "workflow_run_uid") for item in _list(references.get("workflow_run_uids"), "workflow_run_uids", 1)],
|
|
|
+ }
|
|
|
+ snapshot = self.repository.evidence_snapshot(references)
|
|
|
+ quality = snapshot.get("quality") or {}
|
|
|
+ sla_events = snapshot.get("sla_events") or []
|
|
|
+ qualified = (
|
|
|
+ quality.get("status") == "success"
|
|
|
+ and float(quality.get("score", -1)) >= float(product["quality_target"])
|
|
|
+ and sla_events
|
|
|
+ and all(item.get("status") in {"met", "recovered"} for item in sla_events)
|
|
|
+ and snapshot.get("lineage")
|
|
|
+ and all(item.get("parse_status") == "resolved" for item in snapshot["lineage"])
|
|
|
+ and snapshot.get("rule_versions")
|
|
|
+ and all(item.get("status") == "published" for item in snapshot["rule_versions"])
|
|
|
+ and snapshot.get("workflow_runs")
|
|
|
+ and all(item.get("status") == "success" for item in snapshot["workflow_runs"])
|
|
|
+ )
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ certificate_body = {
|
|
|
+ "product_uid": uid,
|
|
|
+ "contract_uid": contract["uid"],
|
|
|
+ "contract_version": contract["current_version"],
|
|
|
+ "approval_task_uid": approval_uid,
|
|
|
+ "evidence_snapshot": snapshot,
|
|
|
+ "status": "qualified" if qualified else "unqualified",
|
|
|
+ }
|
|
|
+ record = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "certificate_code": f"DPCERT-{self.now_factory():%Y%m%d}-{self.uid_factory()[-8:].upper()}",
|
|
|
+ **certificate_body,
|
|
|
+ "content_hash": _canonical_hash(certificate_body),
|
|
|
+ "issued_by": actor,
|
|
|
+ "issued_at": now,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ result = self.repository.create_certificate(record)
|
|
|
+ self.repository.add_event(
|
|
|
+ uid,
|
|
|
+ "certificate_issued",
|
|
|
+ actor,
|
|
|
+ {"certificate_uid": record["uid"], "status": record["status"]},
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def create_feedback(self, product_uid: str, payload: Any, *, actor_uid: str):
|
|
|
+ body = _closed(
|
|
|
+ payload,
|
|
|
+ {"category", "rating", "summary", "details"},
|
|
|
+ "product feedback",
|
|
|
+ )
|
|
|
+ uid = _uid(product_uid, "product_uid")
|
|
|
+ if self.repository.get_product(uid) is None:
|
|
|
+ raise LookupError("governed product was not found")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ if self.repository.users_available({actor}) != {actor}:
|
|
|
+ raise ValueError("feedback author is unavailable")
|
|
|
+ category = _string(body.get("category"), "category", 30)
|
|
|
+ if category not in FEEDBACK_CATEGORIES:
|
|
|
+ raise ValueError("unsupported feedback category")
|
|
|
+ try:
|
|
|
+ rating = int(body.get("rating"))
|
|
|
+ except (TypeError, ValueError) as error:
|
|
|
+ raise ValueError("rating must be an integer") from error
|
|
|
+ if rating < 1 or rating > 5:
|
|
|
+ raise ValueError("rating is outside the allowed range")
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ record = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "product_uid": uid,
|
|
|
+ "category": category,
|
|
|
+ "rating": rating,
|
|
|
+ "summary": _string(body.get("summary"), "summary", 300),
|
|
|
+ "details": _string(body.get("details"), "details", 2000),
|
|
|
+ "status": "open",
|
|
|
+ "assignee_uid": None,
|
|
|
+ "resolution": None,
|
|
|
+ "evidence_refs": [],
|
|
|
+ "current_version": 1,
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now,
|
|
|
+ "updated_by": actor,
|
|
|
+ "updated_at": now,
|
|
|
+ "closed_at": None,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ result = self.repository.create_feedback(record)
|
|
|
+ self.repository.add_event(uid, "feedback_created", actor, {"feedback_uid": record["uid"]})
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def transition_feedback(
|
|
|
+ self, feedback_uid: str, payload: Any, *, expected_version: int, actor_uid: str
|
|
|
+ ):
|
|
|
+ body = _closed(
|
|
|
+ payload,
|
|
|
+ {"action", "assignee_uid", "note", "evidence_refs"},
|
|
|
+ "feedback transition",
|
|
|
+ )
|
|
|
+ uid = _uid(feedback_uid, "feedback_uid")
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ feedback = self.repository.get_feedback(uid)
|
|
|
+ if feedback is None:
|
|
|
+ raise LookupError("product feedback was not found")
|
|
|
+ product = self.repository.get_product(feedback["product_uid"])
|
|
|
+ action = _string(body.get("action"), "action", 30)
|
|
|
+ transitions = {
|
|
|
+ ("open", "triage"): ("triaged", "feedback_triaged"),
|
|
|
+ ("triaged", "start"): ("in_progress", "feedback_started"),
|
|
|
+ ("in_progress", "resolve"): ("resolved", "feedback_resolved"),
|
|
|
+ ("resolved", "close"): ("closed", "feedback_closed"),
|
|
|
+ ("closed", "reopen"): ("open", "feedback_reopened"),
|
|
|
+ }
|
|
|
+ target = transitions.get((feedback["status"], action))
|
|
|
+ if target is None:
|
|
|
+ raise RuntimeError("feedback transition is not allowed")
|
|
|
+ if action in {"triage", "start", "resolve"} and actor not in {
|
|
|
+ product["owner_uid"],
|
|
|
+ feedback.get("assignee_uid"),
|
|
|
+ }:
|
|
|
+ raise PermissionError("only the owner or assignee can improve feedback")
|
|
|
+ if action in {"close", "reopen"} and actor not in {
|
|
|
+ product["owner_uid"],
|
|
|
+ feedback["created_by"],
|
|
|
+ }:
|
|
|
+ raise PermissionError("only the owner or author can confirm feedback")
|
|
|
+ note = _string(body.get("note"), "note", 2000)
|
|
|
+ if action == "triage":
|
|
|
+ assignee = _uid(body.get("assignee_uid"), "assignee_uid")
|
|
|
+ if self.repository.users_available({assignee}) != {assignee}:
|
|
|
+ raise ValueError("feedback assignee is unavailable")
|
|
|
+ feedback["assignee_uid"] = assignee
|
|
|
+ evidence = [
|
|
|
+ _string(item, "evidence reference", 1000)
|
|
|
+ for item in body.get("evidence_refs", [])
|
|
|
+ ]
|
|
|
+ if action == "resolve" and not evidence:
|
|
|
+ raise ValueError("feedback resolution requires evidence")
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ feedback.update(
|
|
|
+ {
|
|
|
+ "status": target[0],
|
|
|
+ "resolution": note if action == "resolve" else feedback.get("resolution"),
|
|
|
+ "evidence_refs": evidence if action == "resolve" else feedback.get("evidence_refs", []),
|
|
|
+ "updated_by": actor,
|
|
|
+ "updated_at": now,
|
|
|
+ "closed_at": now if action == "close" else None,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ result = self.repository.update_feedback(
|
|
|
+ feedback,
|
|
|
+ int(expected_version),
|
|
|
+ target[1],
|
|
|
+ actor,
|
|
|
+ {"note": note, "evidence_refs": evidence},
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def dashboard(self):
|
|
|
+ return self.repository.dashboard()
|