Deprecate kg.ProvenanceTracker in favor of ProvenanceManager

This commit is contained in:
Sameer6305
2026-07-16 22:44:39 +05:30
parent 18c8ba58ef
commit 947ecf186a
4 changed files with 97 additions and 63 deletions
+51
View File
@@ -7,14 +7,23 @@ Tracks the sources and lineage of entities and relationships.
import csv
import io
import json
import warnings
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
_MIGRATION_GUIDE_URL = "docs/migration/kg-provenance-tracker.md"
class ProvenanceTracker:
"""
Tracks provenance (source lineage) for knowledge graph entities.
.. deprecated::
``ProvenanceTracker`` is deprecated in favor of
:class:`semantica.provenance.ProvenanceManager` and will be removed
in a future major version. See the migration guide at
``docs/migration/kg-provenance-tracker.md``.
Usage:
tracker = ProvenanceTracker()
tracker.track_entity("E1", "doc1.txt", metadata={"type": "file"})
@@ -22,6 +31,13 @@ class ProvenanceTracker:
"""
def __init__(self):
warnings.warn(
"ProvenanceTracker is deprecated and will be removed in a future "
"major version. Use semantica.provenance.ProvenanceManager instead. "
f"See migration guide: {_MIGRATION_GUIDE_URL}",
DeprecationWarning,
stacklevel=2,
)
self._records: Dict[str, List[Dict[str, Any]]] = {}
def track_entity(
@@ -31,6 +47,13 @@ class ProvenanceTracker:
metadata: Optional[Dict[str, Any]] = None,
) -> None:
"""Record that entity_id was derived from source."""
warnings.warn(
"ProvenanceTracker.track_entity() is deprecated; use "
"ProvenanceManager.track_entity() instead. "
f"See migration guide: {_MIGRATION_GUIDE_URL}",
DeprecationWarning,
stacklevel=2,
)
if entity_id not in self._records:
self._records[entity_id] = []
entry: Dict[str, Any] = {
@@ -43,6 +66,13 @@ class ProvenanceTracker:
def get_all_sources(self, entity_id: str) -> List[Dict[str, Any]]:
"""Return all provenance records for entity_id."""
warnings.warn(
"ProvenanceTracker.get_all_sources() is deprecated; use "
"ProvenanceManager.get_all_sources() instead. "
f"See migration guide: {_MIGRATION_GUIDE_URL}",
DeprecationWarning,
stacklevel=2,
)
return self._records.get(entity_id, [])
def clear(self, entity_id: Optional[str] = None) -> None:
@@ -66,6 +96,13 @@ class ProvenanceTracker:
Flat list of matching provenance records (each dict includes
the entity_id under the key "entity_id").
"""
warnings.warn(
"ProvenanceTracker.query_recorded_between() is deprecated with no "
"direct ProvenanceManager equivalent yet; see "
f"migration guide: {_MIGRATION_GUIDE_URL}",
DeprecationWarning,
stacklevel=2,
)
start_dt = self._parse_dt(start)
end_dt = self._parse_dt(end)
@@ -93,6 +130,13 @@ class ProvenanceTracker:
Returns an empty list for a fact with no recorded provenance.
"""
warnings.warn(
"ProvenanceTracker.revision_history() is deprecated with no "
"direct ProvenanceManager equivalent yet; see "
f"migration guide: {_MIGRATION_GUIDE_URL}",
DeprecationWarning,
stacklevel=2,
)
records = self._records.get(fact_id, [])
if not records:
return []
@@ -133,6 +177,13 @@ class ProvenanceTracker:
Returns:
String containing the serialized audit log.
"""
warnings.warn(
"ProvenanceTracker.export_audit_log() is deprecated with no "
"direct ProvenanceManager equivalent yet; see "
f"migration guide: {_MIGRATION_GUIDE_URL}",
DeprecationWarning,
stacklevel=2,
)
rows = []
for fact_id in fact_ids:
for entry in self.revision_history(fact_id):
+15 -16
View File
@@ -61,20 +61,17 @@ class TestKGModule:
# Test basic functionality
tracker.track_entity("test_entity", source="test_source")
lineage = tracker.get_lineage("test_entity")
assert lineage is not None
assert "sources" in lineage
# NOTE: tracker.get_lineage() was never implemented on
# kg.ProvenanceTracker; this tested an intended unified-backend
# migration that never happened (#744). ProvenanceTracker is now
# deprecated in favor of semantica.provenance.ProvenanceManager.
def test_kg_uses_unified_backend(self):
"""Test that kg module uses unified backend."""
from semantica.kg import ProvenanceTracker
tracker = ProvenanceTracker()
# Check if using unified backend
assert hasattr(tracker, '_use_unified')
assert hasattr(tracker, '_unified_manager')
# NOTE: test_kg_uses_unified_backend removed; it only asserted the
# presence of _use_unified/_unified_manager attributes, which were
# never implemented on kg.ProvenanceTracker. This tested an intended
# unified-backend migration that never happened (#744).
# ProvenanceTracker is now deprecated in favor of
# semantica.provenance.ProvenanceManager.
def test_kg_graph_builder_ready(self):
"""Test GraphBuilder is ready for provenance."""
@@ -413,6 +410,10 @@ class TestCrossModuleIntegration:
# Track with kg
kg_tracker = KGTracker()
kg_tracker.track_entity("entity_1", source="doc_1")
# NOTE: kg_tracker.get_lineage() was never implemented on
# kg.ProvenanceTracker; this tested an intended unified-backend
# migration that never happened (#744). ProvenanceTracker is now
# deprecated in favor of semantica.provenance.ProvenanceManager.
# Track with split
split_tracker = SplitTracker()
@@ -420,11 +421,9 @@ class TestCrossModuleIntegration:
chunk.id = "chunk_1"
split_tracker.track_chunk(chunk, source_document="doc_1")
# Both should work
kg_lineage = kg_tracker.get_lineage("entity_1")
# split.ProvenanceTracker.get_provenance() is real and still works
split_prov = split_tracker.get_provenance("chunk_1")
assert kg_lineage is not None
assert split_prov is not None
def test_unified_manager_with_all_modules(self):
+27 -41
View File
@@ -21,24 +21,16 @@ class TestKGProvenanceBackwardCompat:
# Track entity (existing API)
tracker.track_entity("entity_1", source="doc_1", metadata={"confidence": 0.9})
# Get lineage (existing API)
lineage = tracker.get_lineage("entity_1")
# Verify existing return format
assert "sources" in lineage
assert "first_seen" in lineage
assert "last_updated" in lineage
assert "metadata" in lineage
# NOTE: tracker.get_lineage() was never implemented on
# kg.ProvenanceTracker; this tested an intended unified-backend
# migration that never happened (#744). ProvenanceTracker is now
# deprecated in favor of semantica.provenance.ProvenanceManager.
def test_track_relationship_unchanged(self):
"""Test relationship tracking works unchanged."""
tracker = KGProvenanceTracker()
tracker.track_relationship("rel_1", source="doc_1", metadata={"type": "founded"})
lineage = tracker.get_lineage("rel_1")
assert lineage is not None
# NOTE: test_track_relationship_unchanged removed; it only exercised
# tracker.track_relationship(), which was never implemented on
# kg.ProvenanceTracker. This tested an intended unified-backend
# migration that never happened (#744). ProvenanceTracker is now
# deprecated in favor of semantica.provenance.ProvenanceManager.
def test_get_all_sources_unchanged(self):
"""Test get_all_sources returns expected format."""
@@ -53,22 +45,14 @@ class TestKGProvenanceBackwardCompat:
assert len(sources) >= 2
for source in sources:
assert "source" in source
assert "timestamp" in source
assert "recorded_at" in source
def test_batch_operations_unchanged(self):
"""Test batch operations work unchanged."""
tracker = KGProvenanceTracker()
entities = [
{"id": "entity_1", "confidence": 0.9},
{"id": "entity_2", "confidence": 0.85}
]
count = tracker.track_entities_batch(entities, "doc_1")
assert count == 2
assert tracker.get_lineage("entity_1") is not None
assert tracker.get_lineage("entity_2") is not None
# NOTE: test_batch_operations_unchanged removed; it only exercised
# tracker.track_entities_batch() and tracker.get_lineage(), neither of
# which was ever implemented on kg.ProvenanceTracker. This tested an
# intended unified-backend migration that never happened (#744).
# ProvenanceTracker is now deprecated in favor of
# semantica.provenance.ProvenanceManager.
class TestSplitProvenanceBackwardCompat:
@@ -196,10 +180,10 @@ class TestGracefulDegradation:
# Should work even if unified backend has issues
tracker.track_entity("entity_1", source="doc_1")
lineage = tracker.get_lineage("entity_1")
assert lineage is not None
assert "sources" in lineage
# NOTE: tracker.get_lineage() was never implemented on
# kg.ProvenanceTracker; this tested an intended unified-backend
# migration that never happened (#744). ProvenanceTracker is now
# deprecated in favor of semantica.provenance.ProvenanceManager.
def test_split_tracker_fallback(self):
"""Test split.ProvenanceTracker falls back to legacy on error."""
@@ -221,18 +205,20 @@ class TestExistingTestsPass:
tracker = KGProvenanceTracker()
# Test 1: Basic tracking
# NOTE: tracker.get_provenance() was never implemented on
# kg.ProvenanceTracker; this tested an intended unified-backend
# migration that never happened (#744). ProvenanceTracker is now
# deprecated in favor of semantica.provenance.ProvenanceManager.
tracker.track_entity("e1", "src1")
assert tracker.get_provenance("e1") is not None
# Test 2: Multiple sources
tracker.track_entity("e1", "src2")
sources = tracker.get_all_sources("e1")
assert len(sources) >= 2
# Test 3: Metadata
tracker.track_entity("e2", "src1", metadata={"key": "value"})
lineage = tracker.get_lineage("e2")
assert "metadata" in lineage
# NOTE: Test 3 (metadata via tracker.get_lineage()) removed; that
# method was never implemented on kg.ProvenanceTracker. This tested
# an intended unified-backend migration that never happened (#744).
def test_split_provenance_existing_behavior(self):
"""Test existing split.ProvenanceTracker behavior is preserved."""
+4 -6
View File
@@ -37,12 +37,10 @@ class TestEndToEndProvenance:
# Track with kg tracker
kg_tracker.track_entity("kg_entity_1", source="kg_doc_1")
# Verify it was tracked
lineage = kg_tracker.get_lineage("kg_entity_1")
assert lineage is not None
assert "sources" in lineage
# NOTE: kg_tracker.get_lineage() was never implemented on
# kg.ProvenanceTracker; this tested an intended unified-backend
# migration that never happened (#744). ProvenanceTracker is now
# deprecated in favor of semantica.provenance.ProvenanceManager.
def test_split_to_unified_integration(self):
"""Test split.ProvenanceTracker uses unified backend."""