orchestrata_ultimate.knowledge_graph.versioning

Reference › orchestrata_ultimate › knowledge_graph › versioning

Git-tag-style named versions of knowledge graph entities (Spec 008, REQ-008-005 and REQ-008-029). Version is a dataclass with a generated UUID, the entity id, a version number starting at 1, a label, a snapshot dictionary, a UTC creation time and a creator. VersioningManager keeps a dictionary from entity id to its list of versions in memory, so nothing is written to the graph or to Redis. create_version appends a snapshot, get_version returns a numbered version or the latest, get_all_versions and get_version_count list them, and clear empties the store. compare_versions diffs the snapshot keys of two versions into added, removed and changed (with old and new values), and returns {"error": "Version not found"} when either version is missing. It complements the graph-backed history in temporal. Only tests import it.

At a glance

Class diagram

Dependency diagram

API

Classes

Version (dataclass)

Decorators: dataclass, traces(spec='008', story='US-006', reqs=['REQ-008-005']) · Line 26

A named version of an entity.

VersioningManager

Decorators: traces(spec='008', story='US-001', reqs=['REQ-008-029']) · Line 50

Manages named versions of entities.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/knowledge_graph/versioning.py

"""Entity versioning for the knowledge graph.

Tracks named versions (like git tags) for entities.
"""

# Spec: 008 - Knowledge Graph

import logging
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)


@traces(spec="008", story="US-006", reqs=["REQ-008-005"])
def _utc_now() -> datetime:
    return datetime.now(UTC)


@dataclass
@traces(spec="008", story="US-006", reqs=["REQ-008-005"])
class Version:
    """A named version of an entity."""

    version_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    entity_id: str = ""
    version_number: int = 1
    label: str = ""
    snapshot: dict[str, Any] = field(default_factory=dict)
    created_at: datetime = field(default_factory=_utc_now)
    created_by: str = ""

    def to_dict(self) -> dict[str, Any]:
        return {
            "version_id": self.version_id,
            "entity_id": self.entity_id,
            "version_number": self.version_number,
            "label": self.label,
            "snapshot": self.snapshot,
            "created_at": self.created_at.isoformat(),
            "created_by": self.created_by,
        }


@traces(spec="008", story="US-001", reqs=["REQ-008-029"])
class VersioningManager:
    """Manages named versions of entities."""

    def __init__(self) -> None:
        self._versions: dict[str, list[Version]] = {}

    @traces(spec="008", story="US-001", reqs=["REQ-008-029"])
    def create_version(
        self,
        entity_id: str,
        snapshot: dict[str, Any],
        label: str = "",
        created_by: str = "",
    ) -> Version:
        """Create a new version of an entity."""
        existing = self._versions.get(entity_id, [])
        version_number = len(existing) + 1

        version = Version(
            entity_id=entity_id,
            version_number=version_number,
            label=label or f"v{version_number}",
            snapshot=dict(snapshot),
            created_by=created_by,
        )
        existing.append(version)
        self._versions[entity_id] = existing
        return version

    @traces(spec="008", story="US-001", reqs=["REQ-008-029"])
    def get_version(
        self, entity_id: str, version_number: int | None = None
    ) -> Version | None:
        """Get a specific version (latest if number not specified)."""
        versions = self._versions.get(entity_id, [])
        if not versions:
            return None
        if version_number is None:
            return versions[-1]
        for v in versions:
            if v.version_number == version_number:
                return v
        return None

    @traces(spec="008", story="US-001", reqs=["REQ-008-029"])
    def get_all_versions(self, entity_id: str) -> list[Version]:
        """Get all versions of an entity."""
        return list(self._versions.get(entity_id, []))

    @traces(spec="008", story="US-001", reqs=["REQ-008-029"])
    def compare_versions(self, entity_id: str, v1: int, v2: int) -> dict[str, Any]:
        """Compare two versions and return differences."""
        ver1 = self.get_version(entity_id, v1)
        ver2 = self.get_version(entity_id, v2)

        if not ver1 or not ver2:
            return {"error": "Version not found"}

        added = {}
        removed = {}
        changed = {}

        all_keys = set(ver1.snapshot.keys()) | set(ver2.snapshot.keys())
        for key in all_keys:
            v1_val = ver1.snapshot.get(key)
            v2_val = ver2.snapshot.get(key)
            if v1_val is None and v2_val is not None:
                added[key] = v2_val
            elif v1_val is not None and v2_val is None:
                removed[key] = v1_val
            elif v1_val != v2_val:
                changed[key] = {"old": v1_val, "new": v2_val}

        return {"added": added, "removed": removed, "changed": changed}

    def get_version_count(self, entity_id: str) -> int:
        """Get number of versions for an entity."""
        return len(self._versions.get(entity_id, []))

    def clear(self) -> None:
        """Clear all version data."""
        self._versions.clear()