semantica/docs/reference/change_management.md

Change Management API Reference

Comprehensive API documentation for the Enhanced Change Management module in Semantica.

Overview

The semantica.change_management module provides enterprise-grade version control, audit trails, and compliance tracking for knowledge graphs and ontologies. It includes persistent storage backends, detailed change tracking, data integrity verification, and standardized metadata structures.

Module Structure

semantica.change_management/
├── change_log.py              # Standardized metadata structures
├── version_storage.py         # Storage abstraction and implementations
├── managers.py                # Enhanced version managers
├── ontology_version_manager.py # Ontology version management
└── change_management_usage.md # Usage guide

Quick Import

from semantica.change_management import (
    # Metadata
    ChangeLogEntry,
    
    # Storage
    VersionStorage,
    InMemoryVersionStorage,
    SQLiteVersionStorage,
    
    # Utilities
    compute_checksum,
    verify_checksum,
    
    # Version Managers
    BaseVersionManager,
    TemporalVersionManager,
    OntologyVersionManager,
    VersionManager,
    OntologyVersion
)

---

Core Classes

ChangeLogEntry

Standardized metadata structure for version changes with validation.

Class Definition

@dataclass
class ChangeLogEntry:
    """
    Standardized change log entry with validation.
    
    Attributes:
        timestamp: ISO 8601 formatted timestamp
        author: Email address of the change author
        description: Change description (max 500 characters)
        change_id: Optional ID linking to external systems
    """
    timestamp: str
    author: str
    description: str
    change_id: Optional[str] = None

Methods

__post_init__()

Validates all fields after initialization.

Raises:

• ValidationError: If any field validation fails

Example:

entry = ChangeLogEntry(
    timestamp="2024-01-30T12:00:00Z",
    author="user@example.com",
    description="Updated entity relationships",
    change_id="TICKET-123"
)

create_now(author, description, change_id=None) (classmethod)

Creates a change log entry with the current timestamp.

Parameters:

• author (str): Email address of the change author
• description (str): Change description (max 500 characters)
• change_id (str, optional): ID linking to external systems

Returns:

• ChangeLogEntry: New instance with current timestamp

Example:

entry = ChangeLogEntry.create_now(
    author="developer@company.com",
    description="Fixed entity resolution bug",
    change_id="JIRA-1234"
)

Validation Rules

• Timestamp: Must be valid ISO 8601 format with 'T' separator
• Author: Must be valid email format (RFC 5322)
• Description: Maximum 500 characters
• Change ID: Optional, no validation

---

VersionStorage

Abstract base class for storage implementations.

Class Definition

class VersionStorage(ABC):
    """
    Abstract base class for version storage backends.
    
    Provides interface for saving, retrieving, and managing version snapshots.
    """

Abstract Methods

save(snapshot)

Save a version snapshot.

Parameters:

• snapshot (Dict[str, Any]): Version snapshot dictionary with metadata

Raises:

• ValidationError: If snapshot data is invalid
• ProcessingError: If save operation fails

Example:

snapshot = {
    "label": "v1.0",
    "timestamp": "2024-01-30T12:00:00Z",
    "author": "user@example.com",
    "description": "Initial version",
    "data": {...}
}
storage.save(snapshot)

get(label)

Retrieve a version snapshot by label.

Parameters:

• label (str): Version label to retrieve

Returns:

• Optional[Dict[str, Any]]: Snapshot dictionary or None if not found

Example:

snapshot = storage.get("v1.0")
if snapshot:
    print(f"Retrieved: {snapshot['label']}")

list_all()

List all version snapshots.

Returns:

• List[Dict[str, Any]]: List of snapshot metadata dictionaries

Example:

versions = storage.list_all()
for v in versions:
    print(f"{v['label']}: {v['description']}")

exists(label)

Check if a version exists.

Parameters:

• label (str): Version label to check

Returns:

• bool: True if version exists, False otherwise

Example:

if storage.exists("v1.0"):
    print("Version exists")

delete(label)

Delete a version snapshot.

Parameters:

• label (str): Version label to delete

Returns:

• bool: True if deleted, False if not found

Example:

if storage.delete("v1.0"):
    print("Version deleted")

---

InMemoryVersionStorage

In-memory version storage implementation.

Class Definition

class InMemoryVersionStorage(VersionStorage):
    """
    In-memory version storage implementation.
    
    Fast, volatile storage for development and testing.
    Data is lost when the process ends.
    """

Constructor

def __init__(self):
    """Initialize in-memory storage."""

Example:

storage = InMemoryVersionStorage()

Performance Characteristics

• Save: 0.37-16ms (10-1000 entities)
• Get: 0.20-16ms (10-1000 entities)
• List: <0.03ms
• Thread-safe: Yes (uses RLock)

Use Cases

• Development and testing
• Temporary version tracking
• High-performance scenarios where persistence is not required

---

SQLiteVersionStorage

SQLite-based persistent version storage implementation.

Class Definition

class SQLiteVersionStorage(VersionStorage):
    """
    SQLite-based persistent version storage implementation.
    
    Provides persistence across process restarts with ACID guarantees.
    """

Constructor

def __init__(self, storage_path: str):
    """
    Initialize SQLite storage.
    
    Args:
        storage_path: Path to SQLite database file
    """

Parameters:

• storage_path (str): Path to SQLite database file (created if doesn't exist)

Example:

storage = SQLiteVersionStorage("versions.db")

Database Schema

CREATE TABLE versions (
    label TEXT PRIMARY KEY,
    timestamp TEXT NOT NULL,
    author TEXT NOT NULL,
    description TEXT,
    checksum TEXT,
    snapshot_data TEXT NOT NULL,
    created_at TEXT NOT NULL
)

Performance Characteristics

• Save: 7-25ms (10-1000 entities)
• Get: 2-8ms (10-1000 entities)
• List: 0.6-13ms
• Thread-safe: Yes (uses RLock)
• ACID: Full transaction support

Use Cases

• Production deployments
• Long-term version storage
• Compliance and audit requirements
• Multi-process environments

---

BaseVersionManager

Abstract base class for version managers.

Class Definition

class BaseVersionManager(ABC):
    """
    Abstract base class for version managers.
    
    Provides common functionality for version management across
    different data types (knowledge graphs, ontologies, etc.).
    """

Constructor

def __init__(self, storage_path: Optional[str] = None):
    """
    Initialize base version manager.
    
    Args:
        storage_path: Path to SQLite database file for persistent storage.
                     If None, uses in-memory storage.
    """

Parameters:

• storage_path (str, optional): Path to SQLite database file

Example:

# In-memory storage
manager = BaseVersionManager()

# Persistent storage
manager = BaseVersionManager(storage_path="versions.db")

Abstract Methods

create_snapshot(data, version_label, author, description, **options)

Create a versioned snapshot of the data.

Parameters:

• data (Any): Data to snapshot
• version_label (str): Version label
• author (str): Email address of the author
• description (str): Change description
• **options: Additional options

Returns:

• Dict[str, Any]: Snapshot with metadata and checksum

compare_versions(version1, version2, **options)

Compare two versions and return detailed differences.

Parameters:

• version1 (Any): First version (label or snapshot)
• version2 (Any): Second version (label or snapshot)
• **options: Comparison options

Returns:

• Dict[str, Any]: Detailed differences

Concrete Methods

list_versions()

List all version snapshots.

Returns:

• List[Dict[str, Any]]: List of version metadata

Example:

versions = manager.list_versions()
for v in versions:
    print(f"{v['label']}: {v['description']}")

get_version(label)

Retrieve specific version by label.

Parameters:

• label (str): Version label

Returns:

• Optional[Dict[str, Any]]: Version snapshot or None

Example:

version = manager.get_version("v1.0")

verify_checksum(snapshot)

Verify data integrity using checksum.

Parameters:

• snapshot (Dict[str, Any]): Snapshot to verify

Returns:

• bool: True if checksum is valid

Example:

is_valid = manager.verify_checksum(snapshot)

---

TemporalVersionManager

Enhanced temporal version management engine for knowledge graphs.

Class Definition

class TemporalVersionManager(BaseVersionManager):
    """
    Enhanced temporal version management engine for knowledge graphs.
    
    Features:
        - Persistent snapshot storage (SQLite or in-memory)
        - Detailed change tracking with entity-level diffs
        - SHA-256 checksums for data integrity
        - Standardized metadata with author attribution
        - Version comparison with backward compatibility
        - Input validation and security features
    """

Constructor

def __init__(self, storage_path: Optional[str] = None, **config):
    """
    Initialize enhanced temporal version manager.
    
    Args:
        storage_path: Path to SQLite database file for persistent storage.
                     If None, uses in-memory storage
        **config: Additional configuration options
    """

Parameters:

• storage_path (str, optional): Path to SQLite database file
• **config: Additional configuration options

Example:

# In-memory storage
manager = TemporalVersionManager()

# Persistent storage
manager = TemporalVersionManager(storage_path="kg_versions.db")

Methods

create_snapshot(graph, version_label, author, description, **options)

Create and store snapshot with checksum and metadata.

Parameters:

• graph (Dict[str, Any]): Knowledge graph dict with "entities" and "relationships"
• version_label (str): Version string (e.g., "v1.0")
• author (str): Email address of the change author
• description (str): Change description (max 500 chars)
• **options: Additional options

Returns:

• Dict[str, Any]: Snapshot with metadata and checksum

Raises:

• ValidationError: If input validation fails
• ProcessingError: If snapshot creation fails

Example:

graph = {
    "entities": [
        {"id": "e1", "name": "Entity 1", "type": "Person"},
        {"id": "e2", "name": "Entity 2", "type": "Organization"}
    ],
    "relationships": [
        {"source": "e1", "target": "e2", "type": "works_for"}
    ]
}

snapshot = manager.create_snapshot(
    graph,
    version_label="v1.0",
    author="user@example.com",
    description="Initial knowledge graph"
)

print(f"Created: {snapshot['label']}")
print(f"Checksum: {snapshot['checksum']}")

compare_versions(version1, version2, **options)

Compare two versions with detailed entity and relationship diffs.

Parameters:

• version1 (Union[str, Dict]): First version (label or snapshot dict)
• version2 (Union[str, Dict]): Second version (label or snapshot dict)
• **options: Comparison options

Returns:

• Dict[str, Any]: Detailed differences including:
  • summary: Aggregate statistics
  • entity_changes: Entity-level changes
  • relationship_changes: Relationship-level changes

Example:

diff = manager.compare_versions("v1.0", "v2.0")

print(f"Entities added: {diff['summary']['entities_added']}")
print(f"Entities modified: {diff['summary']['entities_modified']}")
print(f"Relationships added: {diff['summary']['relationships_added']}")

# Detailed entity changes
for entity_id, changes in diff['entity_changes'].items():
    print(f"Entity {entity_id}: {changes['status']}")
    if changes['status'] == 'modified':
        print(f"  Before: {changes['before']}")
        print(f"  After: {changes['after']}")

Performance

• Snapshot Creation: 1.40-54ms (50-2000 entities)
• Version Retrieval: 0.65-26ms (50-2000 entities)
• Version Comparison: 3.46-33ms (100-1000 entities)
• Concurrent Throughput: 500+ operations/second

---

OntologyVersionManager

Enhanced version management for ontologies.

Class Definition

class OntologyVersionManager(BaseVersionManager):
    """
    Enhanced version management for ontologies.
    
    Features:
        - Persistent ontology snapshot storage
        - Structural comparison (classes, properties, axioms)
        - SHA-256 checksums for data integrity
        - Standardized metadata with author attribution
    """

Constructor

def __init__(self, storage_path: Optional[str] = None, **config):
    """
    Initialize enhanced version manager for ontologies.
    
    Args:
        storage_path: Path to SQLite database file for persistent storage.
                     If None, uses in-memory storage
        **config: Additional configuration options
    """

Example:

manager = OntologyVersionManager(storage_path="ontology_versions.db")

Methods

create_snapshot(ontology, version_label, author, description, **options)

Create ontology snapshot with metadata.

Parameters:

• ontology (Dict[str, Any]): Ontology dict with structure information
• version_label (str): Version label
• author (str): Email address of the author
• description (str): Change description
• **options: Additional options

Returns:

• Dict[str, Any]: Ontology snapshot with metadata

Example:

ontology = {
    "uri": "https://example.com/ontology",
    "version_info": {"version": "1.0", "date": "2024-01-30"},
    "structure": {
        "classes": ["Person", "Organization", "Location"],
        "properties": ["name", "address", "email"],
        "individuals": ["JohnDoe", "ACME_Corp"],
        "axioms": ["Person hasAddress exactly 1 Location"]
    }
}

snapshot = manager.create_snapshot(
    ontology,
    version_label="ont_v1.0",
    author="architect@example.com",
    description="Initial ontology design"
)

compare_versions(version1, version2, **options)

Compare ontology versions with structural analysis.

Parameters:

• version1 (Union[str, Dict]): First version
• version2 (Union[str, Dict]): Second version
• **options: Comparison options

Returns:

• Dict[str, Any]: Structural differences including:
  • classes_added, classes_removed
  • properties_added, properties_removed
  • individuals_added, individuals_removed
  • axioms_added, axioms_removed, axioms_modified

Example:

diff = manager.compare_versions("ont_v1.0", "ont_v2.0")

print(f"Classes added: {diff['classes_added']}")
print(f"Properties added: {diff['properties_added']}")
print(f"Axioms modified: {diff['axioms_modified']}")

---

Utility Functions

compute_checksum

Compute SHA-256 checksum for data integrity.

Function Signature

def compute_checksum(data: Dict[str, Any]) -> str:
    """
    Compute SHA-256 checksum for data.
    
    Args:
        data: Dictionary to compute checksum for
        
    Returns:
        SHA-256 checksum as hexadecimal string
    """

Parameters:

• data (Dict[str, Any]): Dictionary to compute checksum for

Returns:

• str: SHA-256 checksum as hexadecimal string

Example:

from semantica.change_management import compute_checksum

data = {"entities": [...], "relationships": [...]}
checksum = compute_checksum(data)
print(f"Checksum: {checksum}")

Performance: 1.29-110ms (100-10,000 entities)

---

verify_checksum

Verify data integrity using stored checksum.

Function Signature

def verify_checksum(snapshot: Dict[str, Any]) -> bool:
    """
    Verify data integrity using checksum.
    
    Args:
        snapshot: Snapshot dictionary with 'checksum' field
        
    Returns:
        True if checksum is valid, False otherwise
    """

Parameters:

• snapshot (Dict[str, Any]): Snapshot dictionary with 'checksum' field

Returns:

• bool: True if checksum is valid, False otherwise

Example:

from semantica.change_management import verify_checksum

snapshot = manager.get_version("v1.0")
is_valid = verify_checksum(snapshot)

if not is_valid:
    print("WARNING: Data integrity compromised!")

Performance: 0.82-96ms (100-10,000 entities)

---

Legacy Classes

VersionManager

Original ontology version manager (moved from semantica.ontology).

Import

from semantica.change_management import VersionManager, OntologyVersion

Note: This class is maintained for backward compatibility. New projects should use OntologyVersionManager.

---

Error Handling

ValidationError

Raised when input validation fails.

Common Causes:

• Invalid email format
• Description exceeds 500 characters
• Invalid ISO 8601 timestamp
• Missing required fields

Example:

from semantica.utils.exceptions import ValidationError

try:
    entry = ChangeLogEntry(
        timestamp="invalid",
        author="not-an-email",
        description="x" * 501
    )
except ValidationError as e:
    print(f"Validation failed: {e}")

ProcessingError

Raised when operations fail.

Common Causes:

• Database connection issues
• File system errors
• Concurrent modification conflicts

Example:

from semantica.utils.exceptions import ProcessingError

try:
    storage.save(snapshot)
except ProcessingError as e:
    print(f"Save failed: {e}")

---

Performance Considerations

Benchmarks

Based on comprehensive performance testing:

Component	Small (100)	Medium (500)	Large (2000)
Snapshot Creation	2.33ms	10.70ms	54.23ms
Version Retrieval	1.88ms	7.33ms	26.04ms
Version Comparison	3.46ms	17.39ms	32.83ms
Checksum Compute	1.29ms	5.48ms	22.15ms
SQLite Save	8.69ms	13.37ms	25.33ms
InMemory Save	1.18ms	10.60ms	14.11ms

Optimization Tips

1. Use appropriate storage backend:

   • Development: InMemoryVersionStorage
   • Production: SQLiteVersionStorage

2. Batch operations when possible:

   for data in batch:
       manager.create_snapshot(data, ...)
   

3. Implement retention policies:

   # Delete old versions periodically
   for version in old_versions:
       storage.delete(version['label'])
   

4. Use concurrent operations:

   • Thread-safe: 500+ operations/second
   • No performance degradation under load

---

Compliance Features

HIPAA Compliance

• Complete audit trails with author attribution
• Timestamp tracking for all changes
• Data integrity verification with checksums
• Secure storage with access controls

SOX Compliance

• Immutable change records
• Detailed change descriptions
• External system linking (change IDs)
• Comprehensive audit reports

FDA 21 CFR Part 11

• Electronic signatures (author email)
• Data integrity verification
• Audit trail generation
• Tamper detection

---

Examples

Complete Healthcare Example

from semantica.change_management import TemporalVersionManager

# Initialize with HIPAA-compliant storage
manager = TemporalVersionManager(storage_path="hipaa_records.db")

# Patient knowledge graph
patient_kg = {
    "entities": [
        {"id": "patient_001", "type": "Patient", "name": "Jane Smith"},
        {"id": "diagnosis_001", "type": "Diagnosis", "code": "I10"}
    ],
    "relationships": [
        {"source": "patient_001", "target": "diagnosis_001", "type": "has_diagnosis"}
    ]
}

# Create versioned record
snapshot = manager.create_snapshot(
    patient_kg,
    "patient_001_v1.0",
    "dr.williams@hospital.com",
    "Initial diagnosis - Essential hypertension"
)

# Verify integrity
assert manager.verify_checksum(snapshot), "Data integrity check failed"

# Generate audit report
for version in manager.list_versions():
    print(f"{version['timestamp']}: {version['label']} by {version['author']}")

---

See Also

• Usage Guide: semantica/change_management/change_management_usage.md
• Performance Tests: tests/change_management/test_performance.py
• CHANGELOG: CHANGELOG.md
• GitHub: https://github.com/Hawksight-AI/semantica