orchestrata_ultimate.knowledge_graph.history

Reference › orchestrata_ultimate › knowledge_graph › history

HistoryManager records and queries a change log for knowledge graph entities (Spec 008, REQ-008-030 and REQ-008-034). record_change builds a ChangeEntry (entity id and type, a ChangeType, optional old and new values, who changed it, and the session id) and appends it to an in-memory list plus per-entity and per-session indexes. The constructor accepts an optional Redis client, but the recording path shown here writes only to those in-process structures, so the log lasts as long as the process. Queries are get_change_log (filter by entity, entity type or change type, with a limit of 100 by default), get_timeline (a time range), get_session_changes and get_entity_summary; total_entries and clear complete the surface. It is a separate, lighter record than the versions kept by temporal. Types come from types and types_enums. Only tests import it.

At a glance

Class diagram

Dependency diagram

API

Classes

HistoryManager

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

Records and queries change history.

Functions

Source

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

# mypy: ignore-errors
"""Change history for the knowledge graph.

Records and queries changes to knowledge graph entities
with support for timeline queries and filtering.
"""

# Spec: 008 - Knowledge Graph

from __future__ import annotations

import logging
from collections import defaultdict
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.types import ChangeEntry, ChangeType

if TYPE_CHECKING:
    import redis

logger = logging.getLogger(__name__)


def _utc_now() -> datetime:
    return datetime.now(UTC)


@traces(spec="008", story="US-001", reqs=["REQ-008-030", "REQ-008-034"])
class HistoryManager:
    """Records and queries change history."""

    def __init__(self, redis: redis.Redis | None = None) -> None:
        self._redis = redis
        self._entries: list[ChangeEntry] = []
        self._by_entity: dict[str, list[ChangeEntry]] = defaultdict(list)
        self._by_session: dict[str, list[ChangeEntry]] = defaultdict(list)

    @traces(spec="008", story="US-001", reqs=["REQ-008-030", "REQ-008-034"])
    def record_change(
        self,
        entity_id: str,
        entity_type: str,
        change_type: ChangeType,
        old_value: dict[str, Any] | None = None,
        new_value: dict[str, Any] | None = None,
        changed_by: str = "",
        session_id: str = "",
    ) -> ChangeEntry:
        """Record a change to an entity."""
        entry = ChangeEntry(
            entity_id=entity_id,
            entity_type=entity_type,
            change_type=change_type,
            old_value=old_value,
            new_value=new_value,
            changed_by=changed_by,
            session_id=session_id,
        )
        self._entries.append(entry)
        self._by_entity[entity_id].append(entry)
        if session_id:
            self._by_session[session_id].append(entry)
        return entry

    @traces(spec="008", story="US-001", reqs=["REQ-008-030"])
    def get_change_log(
        self,
        entity_id: str | None = None,
        entity_type: str | None = None,
        change_type: ChangeType | None = None,
        limit: int = 100,
    ) -> list[ChangeEntry]:
        """Get change log with optional filters."""
        if entity_id:
            entries = self._by_entity.get(entity_id, [])
        else:
            entries = self._entries

        filtered = entries
        if entity_type:
            filtered = [e for e in filtered if e.entity_type == entity_type]
        if change_type:
            filtered = [e for e in filtered if e.change_type == change_type]

        return filtered[-limit:]

    @traces(spec="008", story="US-001", reqs=["REQ-008-028"])
    def get_timeline(
        self,
        start: datetime | None = None,
        end: datetime | None = None,
        limit: int = 100,
    ) -> list[ChangeEntry]:
        """Get changes within a time range."""
        entries = self._entries
        if start:
            entries = [e for e in entries if e.changed_at >= start]
        if end:
            entries = [e for e in entries if e.changed_at <= end]
        return entries[-limit:]

    @traces(spec="008", story="US-001", reqs=["REQ-008-043"])
    def get_session_changes(self, session_id: str) -> list[ChangeEntry]:
        """Get all changes in a session."""
        return list(self._by_session.get(session_id, []))

    def get_entity_summary(self, entity_id: str) -> dict[str, Any]:
        """Get a summary of changes for an entity."""
        entries = self._by_entity.get(entity_id, [])
        if not entries:
            return {"entity_id": entity_id, "total_changes": 0}

        type_counts: dict[str, int] = defaultdict(int)
        for entry in entries:
            type_counts[entry.change_type.value] += 1

        return {
            "entity_id": entity_id,
            "total_changes": len(entries),
            "first_change": entries[0].changed_at.isoformat(),
            "last_change": entries[-1].changed_at.isoformat(),
            "change_types": dict(type_counts),
        }

    @property
    def total_entries(self) -> int:
        return len(self._entries)

    def clear(self) -> None:
        """Clear all history."""
        self._entries.clear()
        self._by_entity.clear()
        self._by_session.clear()