"""Generic governance object contracts and versioned domain templates.""" from __future__ import annotations import hashlib import json import re import uuid from collections.abc import Callable from copy import deepcopy from dataclasses import dataclass from typing import Any from app.core.common.identifiers import new_governance_uid IDENTIFIER_RE = re.compile(r"^[a-z][a-z0-9_]{1,63}$") CODE_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]{1,79}$") LIFECYCLE_STATUSES = frozenset({"draft", "active", "retired"}) FIELD_TYPES = frozenset( {"string", "integer", "number", "boolean", "date", "datetime", "object", "array"} ) SECRET_MARKERS = ("password", "secret", "token", "credential", "api_key", "private_key") MAX_OBJECT_TYPES = 100 MAX_COLLECTION_ITEMS = 500 class DomainTemplateValidationError(ValueError): pass class DomainTemplateNotFound(LookupError): pass @dataclass(frozen=True) class GovernanceObjectType: """Runtime contract for one template-defined governance object type.""" template_code: str type_code: str name: str stable_uid_prefix: str source_identity_fields: tuple[str, ...] fields: tuple[dict[str, Any], ...] lifecycle_status: str = "active" current_version: int = 1 @classmethod def from_definition( cls, template_code: str, definition: dict[str, Any], *, current_version: int = 1, ): normalized = _normalize_object_types([definition])[0] return cls( template_code=_require_identifier(template_code, "template_code"), type_code=normalized["type_code"], name=normalized["name"], stable_uid_prefix=normalized["stable_uid_prefix"], source_identity_fields=tuple(normalized["source_identity_fields"]), fields=tuple(normalized["fields"]), lifecycle_status=normalized["lifecycle_status"], current_version=int(current_version), ) def stable_uid(self, source_identity: dict[str, Any]) -> str: missing = [ field for field in self.source_identity_fields if source_identity.get(field) is None or str(source_identity.get(field)).strip() == "" ] if missing: raise DomainTemplateValidationError( "source_identity is missing required fields: " + ", ".join(missing) ) canonical_identity = { field: source_identity[field] for field in self.source_identity_fields } return stable_governance_object_uid( self.template_code, self.type_code, canonical_identity, ) def _canonical_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def _content_hash(value: Any) -> str: return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest() def _require_identifier(value: Any, field: str) -> str: normalized = str(value or "").strip() if not IDENTIFIER_RE.fullmatch(normalized): raise DomainTemplateValidationError( f"{field} must be a lowercase snake_case identifier" ) return normalized def _require_code(value: Any, field: str) -> str: normalized = str(value or "").strip() if not CODE_RE.fullmatch(normalized): raise DomainTemplateValidationError(f"{field} must be a stable code") return normalized def _require_text(value: Any, field: str, max_length: int = 200) -> str: normalized = str(value or "").strip() if not normalized or len(normalized) > max_length: raise DomainTemplateValidationError( f"{field} must be between 1 and {max_length} characters" ) return normalized def _reject_secrets(value: Any, path: str = "template") -> None: if isinstance(value, dict): for key, nested in value.items(): normalized_key = str(key).lower().replace("-", "_") if any(marker in normalized_key for marker in SECRET_MARKERS): raise DomainTemplateValidationError( 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 _normalize_fields(fields: Any, object_code: str) -> list[dict[str, Any]]: if fields is None: fields = [] if not isinstance(fields, list) or len(fields) > MAX_COLLECTION_ITEMS: raise DomainTemplateValidationError(f"{object_code}.fields must be a bounded list") normalized = [] seen = set() for item in fields: if not isinstance(item, dict): raise DomainTemplateValidationError(f"{object_code}.fields items must be objects") code = _require_identifier(item.get("code"), f"{object_code}.field.code") if any(marker in code for marker in SECRET_MARKERS): raise DomainTemplateValidationError( f"secret-bearing field is not allowed at {object_code}.{code}" ) if code in seen: raise DomainTemplateValidationError(f"duplicate field {object_code}.{code}") seen.add(code) field_type = str(item.get("type") or "string").strip().lower() if field_type not in FIELD_TYPES: raise DomainTemplateValidationError( f"{object_code}.{code}.type is unsupported" ) normalized.append( { "code": code, "name": _require_text( item.get("name") or code, f"{object_code}.{code}.name" ), "type": field_type, "required": bool(item.get("required", False)), "description": str(item.get("description") or "").strip(), } ) return sorted(normalized, key=lambda item: item["code"]) def _normalize_object_types(items: Any) -> list[dict[str, Any]]: if not isinstance(items, list) or not items or len(items) > MAX_OBJECT_TYPES: raise DomainTemplateValidationError("object_types must be a non-empty bounded list") normalized = [] seen = set() for item in items: if not isinstance(item, dict): raise DomainTemplateValidationError("object type must be an object") code = _require_identifier(item.get("type_code"), "object_types.type_code") if code in seen: raise DomainTemplateValidationError(f"duplicate object type: {code}") seen.add(code) source_identity_fields = item.get("source_identity_fields") if not isinstance(source_identity_fields, list) or not source_identity_fields: raise DomainTemplateValidationError( f"{code}.source_identity_fields must be a non-empty list" ) source_identity_fields = [ _require_identifier(value, f"{code}.source_identity_fields") for value in source_identity_fields ] if any( marker in field for field in source_identity_fields for marker in SECRET_MARKERS ): raise DomainTemplateValidationError( f"{code}.source_identity_fields cannot contain secret-bearing fields" ) if len(set(source_identity_fields)) != len(source_identity_fields): raise DomainTemplateValidationError( f"{code}.source_identity_fields contains duplicates" ) normalized.append( { "type_code": code, "name": _require_text(item.get("name"), f"{code}.name"), "description": str(item.get("description") or "").strip(), "stable_uid_prefix": _require_text( item.get("stable_uid_prefix"), f"{code}.stable_uid_prefix", 16 ).upper(), "source_identity_fields": source_identity_fields, "fields": _normalize_fields(item.get("fields"), code), "lifecycle_status": str( item.get("lifecycle_status") or "active" ).strip().lower(), "extension": deepcopy(item.get("extension") or {}), } ) if normalized[-1]["lifecycle_status"] not in LIFECYCLE_STATUSES: raise DomainTemplateValidationError( f"{code}.lifecycle_status is unsupported" ) return sorted(normalized, key=lambda item: item["type_code"]) def _normalize_named_collection(value: Any, name: str) -> list[dict[str, Any]]: if value is None: return [] if not isinstance(value, list) or len(value) > MAX_COLLECTION_ITEMS: raise DomainTemplateValidationError(f"{name} must be a bounded list") normalized = [] seen = set() for item in value: if not isinstance(item, dict): raise DomainTemplateValidationError(f"{name} items must be objects") code = _require_code(item.get("code"), f"{name}.code") if code in seen: raise DomainTemplateValidationError(f"duplicate {name} code: {code}") seen.add(code) normalized.append({**deepcopy(item), "code": code}) return sorted(normalized, key=lambda item: item["code"]) def normalize_domain_template(definition: Any) -> dict[str, Any]: if not isinstance(definition, dict): raise DomainTemplateValidationError("template definition must be an object") _reject_secrets(definition) lifecycle_status = str(definition.get("lifecycle_status") or "draft").strip().lower() if lifecycle_status not in LIFECYCLE_STATUSES: raise DomainTemplateValidationError("lifecycle_status is unsupported") seed_data = deepcopy(definition.get("seed_data") or []) if not isinstance(seed_data, list) or len(seed_data) > MAX_COLLECTION_ITEMS: raise DomainTemplateValidationError("seed_data must be a bounded list") normalized = { "template_code": _require_identifier( definition.get("template_code"), "template_code" ), "name": _require_text(definition.get("name"), "name"), "description": str(definition.get("description") or "").strip(), "lifecycle_status": lifecycle_status, "object_types": _normalize_object_types(definition.get("object_types")), "responsibility_roles": _normalize_named_collection( definition.get("responsibility_roles"), "responsibility_roles" ), "rules": _normalize_named_collection(definition.get("rules"), "rules"), "metrics": _normalize_named_collection(definition.get("metrics"), "metrics"), "seed_data": seed_data, "source_identity_contract": { "strategy": "uuid5", "canonicalization": "sorted-json", "required": True, }, "version_contract": { "strategy": "append-only", "current_version_field": "current_version", "content_hash": "sha256", }, "lifecycle_contract": { "statuses": sorted(LIFECYCLE_STATUSES), "transitions": { "draft": ["active", "retired"], "active": ["retired"], "retired": [], }, }, } normalized["content_hash"] = _content_hash( {key: value for key, value in normalized.items() if key != "content_hash"} ) return normalized def stable_governance_object_uid( template_code: str, object_type: str, source_identity: dict[str, Any], ) -> str: template_code = _require_identifier(template_code, "template_code") object_type = _require_identifier(object_type, "object_type") if not isinstance(source_identity, dict) or not source_identity: raise DomainTemplateValidationError("source_identity must be a non-empty object") if any(value is None or str(value).strip() == "" for value in source_identity.values()): raise DomainTemplateValidationError("source_identity values must be non-empty") identity = f"{template_code}/{object_type}/{_canonical_json(source_identity)}" return str(uuid.uuid5(uuid.NAMESPACE_URL, f"dataops-governance:{identity}")) def diff_domain_templates( before: dict[str, Any] | None, after: dict[str, Any], ) -> dict[str, Any]: before = before or {} before_types = { item["type_code"]: item for item in before.get("object_types", []) } after_types = {item["type_code"]: item for item in after.get("object_types", [])} shared = set(before_types) & set(after_types) sections = ("responsibility_roles", "rules", "metrics", "seed_data") return { "added_object_types": sorted(set(after_types) - set(before_types)), "removed_object_types": sorted(set(before_types) - set(after_types)), "changed_object_types": sorted( code for code in shared if before_types[code] != after_types[code] ), "changed_sections": sorted( section for section in sections if before.get(section, []) != after.get(section, []) ), "lifecycle_changed": ( bool(before) and before.get("lifecycle_status") != after.get("lifecycle_status") ), } class DomainTemplateService: def __init__( self, repository, *, uid_factory: Callable[[], str] = new_governance_uid, ): self.repository = repository self.uid_factory = uid_factory def list_templates(self): return self.repository.list_templates() def get_template(self, template_code: str): template_code = _require_identifier(template_code, "template_code") result = self.repository.get_template(template_code) if result is None: raise DomainTemplateNotFound(f"domain template {template_code} was not found") return result def list_imports(self, template_code: str): template_code = _require_identifier(template_code, "template_code") return self.repository.list_imports(template_code) def dry_run(self, definition: Any): normalized = normalize_domain_template(definition) before = self.repository.get_template(normalized["template_code"]) return { "valid": True, "template": normalized, "diff": diff_domain_templates(before, normalized), } def import_template( self, definition: Any, *, actor_uid: str, operation: str = "import", target_version: int | None = None, ): normalized = normalize_domain_template(definition) before = self.repository.get_template(normalized["template_code"]) version = int(before.get("current_version", 0)) + 1 if before else 1 persisted = {**normalized, "current_version": version} audit = { "uid": self.uid_factory(), "template_code": normalized["template_code"], "operation": operation, "status": "applied", "version": version, "target_version": target_version, "before_state": deepcopy(before), "after_state": deepcopy(persisted), "diff": diff_domain_templates(before, normalized), "actor_uid": str(actor_uid), } return self.repository.apply_import( { "uid": self.uid_factory(), "template_code": normalized["template_code"], "version": version, "template": persisted, "audit": audit, } ) def rollback( self, template_code: str, *, target_version: int, actor_uid: str, ): template_code = _require_identifier(template_code, "template_code") if not isinstance(target_version, int) or target_version < 1: raise DomainTemplateValidationError("target_version must be a positive integer") target = self.repository.get_version(template_code, target_version) if target is None: raise DomainTemplateNotFound( f"domain template {template_code} version {target_version} was not found" ) definition = { key: deepcopy(value) for key, value in target.items() if key not in {"current_version", "content_hash"} } return self.import_template( definition, actor_uid=actor_uid, operation="rollback", target_version=target_version, )