Add approval template revision administration

2026-08-04 01:04:39 +02:00
parent 3abbe117e5
commit 7f64247625
15 changed files with 833 additions and 36 deletions
@@ -238,6 +238,75 @@ class SqlApprovalRequests:
            .all()
        )

+    def template_history(
+        self,
+        session: object,
+        principal: object,
+        *,
+        template_id: str,
+    ) -> tuple[Mapping[str, object], ...]:
+        rows = (
+            _session(session)
+            .query(ApprovalTemplateRevision)
+            .filter(
+                ApprovalTemplateRevision.tenant_id == _tenant(principal),
+                ApprovalTemplateRevision.template_id == template_id,
+            )
+            .order_by(ApprovalTemplateRevision.revision.desc())
+            .all()
+        )
+        if not rows:
+            raise LookupError("Approval template not found.")
+        return tuple(_template_mapping(row) for row in rows)
+
+    def compare_template_revisions(
+        self,
+        session: object,
+        principal: object,
+        *,
+        template_id: str,
+        from_revision: int,
+        to_revision: int,
+    ) -> Mapping[str, object]:
+        if from_revision < 1 or to_revision < 1:
+            raise ApprovalStoreError(
+                "Approval template revisions must be positive integers."
+            )
+        rows = (
+            _session(session)
+            .query(ApprovalTemplateRevision)
+            .filter(
+                ApprovalTemplateRevision.tenant_id == _tenant(principal),
+                ApprovalTemplateRevision.template_id == template_id,
+                ApprovalTemplateRevision.revision.in_(
+                    (from_revision, to_revision)
+                ),
+            )
+            .all()
+        )
+        by_revision = {row.revision: row for row in rows}
+        missing = [
+            revision
+            for revision in (from_revision, to_revision)
+            if revision not in by_revision
+        ]
+        if missing:
+            raise LookupError(
+                "Approval template revision not found: "
+                + ", ".join(str(revision) for revision in missing)
+            )
+        before = by_revision[from_revision]
+        after = by_revision[to_revision]
+        return {
+            "template_id": template_id,
+            "from_revision": _template_mapping(before),
+            "to_revision": _template_mapping(after),
+            "changes": _structured_changes(
+                before.payload,
+                after.payload,
+            ),
+        }
+
    def create_request(
        self,
        session: object,
@@ -1073,10 +1142,80 @@ def _template_mapping(row: ApprovalTemplateRevision) -> dict[str, object]:
        "state": row.state,
        "content_sha256": row.content_sha256,
        "recorded_at": _iso(row.recorded_at),
+        "superseded_at": _iso(row.superseded_at),
+        "previous_revision_id": row.previous_revision_id,
+        "actor_id": row.actor_id,
        **dict(row.payload),
    }


+_MISSING = object()
+
+
+def _structured_changes(
+    before: object,
+    after: object,
+    *,
+    path: str = "",
+) -> list[dict[str, object]]:
+    if isinstance(before, Mapping) and isinstance(after, Mapping):
+        changes: list[dict[str, object]] = []
+        keys = sorted(set(before).union(after), key=str)
+        for key in keys:
+            child_path = f"{path}/{_json_pointer_segment(str(key))}"
+            changes.extend(
+                _structured_changes(
+                    before.get(key, _MISSING),
+                    after.get(key, _MISSING),
+                    path=child_path,
+                )
+            )
+        return changes
+    if isinstance(before, list) and isinstance(after, list):
+        changes = []
+        for index in range(max(len(before), len(after))):
+            changes.extend(
+                _structured_changes(
+                    before[index] if index < len(before) else _MISSING,
+                    after[index] if index < len(after) else _MISSING,
+                    path=f"{path}/{index}",
+                )
+            )
+        return changes
+    if before is _MISSING:
+        return [
+            {
+                "path": path or "/",
+                "change": "added",
+                "before": None,
+                "after": after,
+            }
+        ]
+    if after is _MISSING:
+        return [
+            {
+                "path": path or "/",
+                "change": "removed",
+                "before": before,
+                "after": None,
+            }
+        ]
+    if before != after:
+        return [
+            {
+                "path": path or "/",
+                "change": "changed",
+                "before": before,
+                "after": after,
+            }
+        ]
+    return []
+
+
+def _json_pointer_segment(value: str) -> str:
+    return value.replace("~", "~0").replace("/", "~1")
+
+
 def _template_ref(row: ApprovalTemplateRevision) -> ApprovalTemplateRef:
    return ApprovalTemplateRef(
        id=row.template_id,