DataOps-platform/app/core/knowledge/diff.py

from __future__ import annotations

from app.core.knowledge.contracts import (
    KnowledgeSnapshot,
    PointChange,
    SnapshotDiff,
)


def diff_snapshots(old: KnowledgeSnapshot, new: KnowledgeSnapshot) -> SnapshotDiff:
    if (old.source_type, old.source_uid) != (new.source_type, new.source_uid):
        raise ValueError("knowledge snapshots must describe the same source object")

    old_points = {point.point_key: point for point in old.points}
    new_points = {point.point_key: point for point in new.points}
    groups: dict[str, list[PointChange]] = {
        "added": [],
        "modified": [],
        "deleted": [],
        "unchanged": [],
    }
    for point_key in sorted(set(old_points) | set(new_points)):
        old_point = old_points.get(point_key)
        new_point = new_points.get(point_key)
        if old_point is None:
            kind = "added"
        elif new_point is None:
            kind = "deleted"
        elif (
            old_point.content_hash,
            old_point.metadata_hash,
            old_point.permission_hash,
        ) == (
            new_point.content_hash,
            new_point.metadata_hash,
            new_point.permission_hash,
        ):
            kind = "unchanged"
        else:
            kind = "modified"
        groups[kind].append(
            PointChange(
                point_key=point_key,
                change_kind=kind,
                old=old_point,
                new=new_point,
                permission_changed=bool(
                    old_point
                    and new_point
                    and old_point.permission_hash != new_point.permission_hash
                ),
            )
        )
    return SnapshotDiff(
        added=tuple(groups["added"]),
        modified=tuple(groups["modified"]),
        deleted=tuple(groups["deleted"]),
        unchanged=tuple(groups["unchanged"]),
    )