- graph_store: remove create_constraint(), add_nodes_bulk(), add_edges_bulk() → create_nodes(), add_edges() - deduplication: fix PropertyMergeRule → MergeStrategy enum; add_rule() → add_property_rule(); merge() → merge_entities(); remove non-existent UNION/MAX/MIN/VOTING constants - conflicts: set_credibility() → set_source_credibility(); group_by_severity/identify_patterns/analyze_sources → analyze_conflicts() dict keys; generate() → generate_guide(); remove time_window= param from analyze_trends() - reasoning: infer() → forward_chain(); remove apply_transitivity/symmetry/inverse() templates that don't exist; GraphReasoner(kg) → GraphReasoner(); infer(kg) → reason(graph, query) - split: split_document() (singular) → split_documents([parsed]) throughout - seed: remove register_source_object(), populate(), inject(), load_from_file(), diff_versions(), get_version(tag=) — replace with register_source() and load_from_csv/json() - change_management: remove rollback(), get_log_entry(), export_audit_trail(), get_audit_trail() — replace audit section with list_versions() + diff() pattern - export: export_to_file() → export_to_rdf(); YAMLExporter → SemanticNetworkYAMLExporter
13 KiB
title, description, icon
| title | description | icon |
|---|---|---|
| Change Management Module | Version control, SHA-256 checksums, diff analysis, rollback, and audit trails for knowledge graphs and ontologies. | clock-rotate-left |
semantica.change_management provides enterprise-grade versioning and audit trails for knowledge graphs and ontologies. Every snapshot carries a SHA-256 checksum, every modification is logged, and every state can be diffed or rolled back — giving you a complete, tamper-evident record suitable for regulated industries.
What You Get
Snapshot, diff, rollback, and per-entity audit trail for knowledge graphs. Version control for OWL ontologies with diff and schema migration support. Pluggable backends — `InMemoryVersionStorage` for tests, `SQLiteVersionStorage` for production. SHA-256 / SHA-512 checksums to detect any unauthorised graph modification. Structured record of every change: author, timestamp, checksum, and change list. Full tamper-evident version history via `list_versions()` and `diff()` for regulatory review.Typical Workflow
```python from semantica.change_management import TemporalVersionManagermanager = TemporalVersionManager(storage_path="versions.db")
```
for change in diff.changes:
print(f" [{change.type}] {change.element}: {change.description}")
```
TemporalVersionManager
Version control for knowledge graphs — snapshot, diff, and rollback.
Constructor Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
storage_path |
str |
None |
Path to SQLite database; uses in-memory if omitted |
storage |
VersionStorage |
None |
Explicit storage backend instance — overrides storage_path |
List and Retrieve
# List all versions
versions = manager.list_versions()
for v in versions:
print(f"{v.version} — {v.author} — {v.created_at} — {v.checksum[:8]}...")
# Retrieve a specific version
kg_v1 = manager.get_version("v1.0")
Diff Analysis
Compare any two snapshots to see exactly what changed — useful for code review, incident investigation, and regulatory audit:
diff = manager.diff("v1.0", "v2.0")
print(f"Added nodes: {len(diff.added_nodes)}")
print(f"Removed nodes: {len(diff.removed_nodes)}")
print(f"Modified nodes: {len(diff.modified_nodes)}")
print(f"Added edges: {len(diff.added_edges)}")
print(f"Removed edges: {len(diff.removed_edges)}")
print(f"Modified edges: {len(diff.modified_edges)}")
for change in diff.changes:
print(f" [{change.type}] {change.element}: {change.description}")
@dataclass
class DiffResult:
from_version: str # source snapshot ID
to_version: str # target snapshot ID
added_nodes: List[str] # IDs of newly added entities
removed_nodes: List[str] # IDs of deleted entities
modified_nodes: List[str] # IDs of entities with changed properties
added_edges: List[str] # IDs of newly added relationships
removed_edges: List[str] # IDs of deleted relationships
modified_edges: List[str] # IDs of relationships with changed properties
changes: List[ChangeRecord] # ordered list of all individual changes
summary: str # human-readable summary line
OntologyVersionManager
Version control for OWL ontologies — save, diff, and track schema migrations:
from semantica.change_management import OntologyVersionManager, OntologyVersion
manager = OntologyVersionManager()
# Save a version
version: OntologyVersion = manager.save_version(
ontology=ontology,
version="1.2.0",
author="ontology-team",
message="Added FHIR alignment mappings"
)
# Diff two ontology versions
diff = manager.diff("1.1.0", "1.2.0")
for change in diff.changes:
print(f"[{change.type}] {change.class_name}: {change.description}")
VersionStorage Backends
```python from semantica.change_management import SQLiteVersionStorage, TemporalVersionManagerstorage = SQLiteVersionStorage(db_path="versions.db")
manager = TemporalVersionManager(storage=storage)
```
Persists all version history to disk. Survives process restarts. Recommended for any environment where you need to retain the audit trail.
You can also pass the path directly to `TemporalVersionManager`:
```python
manager = TemporalVersionManager(storage_path="versions.db")
```
storage = InMemoryVersionStorage()
manager = TemporalVersionManager(storage=storage)
```
Fast and zero-setup. Data is **not persisted** — all version history is lost when the process exits. Use this for unit tests and development only.
Integrity Verification
SHA-256 checksums detect any unauthorized modification to a graph between snapshots:
from semantica.change_management import compute_checksum, verify_checksum
# Compute checksum for a graph
checksum = compute_checksum(kg)
# Verify graph against a stored checksum
is_valid = verify_checksum(kg, expected_checksum=checksum)
if not is_valid:
raise RuntimeError("Graph has been modified since the checksum was recorded")
ChangeLogEntry
Every version snapshot includes a structured ChangeLogEntry that records the full context of a change:
# Retrieve a version entry
entry = manager.get_version("v1.0")
print(entry.version) # "v1.0"
print(entry.author) # "user@example.com"
print(entry.message) # "Initial knowledge graph"
print(entry.checksum) # SHA-256 hex digest of the full graph state
print(entry.created_at) # datetime of snapshot creation
print(entry.node_count) # total nodes at this snapshot
print(entry.edge_count) # total edges at this snapshot
print(entry.changes) # list[ChangeRecord] — individual property-level changes
@dataclass
class ChangeLogEntry:
snapshot_id: str # unique snapshot identifier
version: str # human-assigned version tag, e.g. "v1.0"
author: str # identity of the user or process that created it
message: str # commit-style description of what changed
checksum: str # SHA-256 hex digest — changes if graph is tampered
created_at: datetime # UTC timestamp of snapshot creation
node_count: int # total entity count at this point in time
edge_count: int # total relationship count at this point in time
changes: List[ChangeRecord] # granular per-property change records
metadata: Dict # arbitrary key-value pairs for custom tagging
Compliance and Version History
All version snapshots form a tamper-evident audit trail. Use list_versions() and diff() to reconstruct and review changes for regulatory purposes:
from semantica.change_management import TemporalVersionManager
manager = TemporalVersionManager(storage_path="versions.db")
# Enumerate the full version history
for entry in manager.list_versions():
print(f"{entry.created_at.isoformat()} | {entry.author} | {entry.version} | {entry.message}")
# Diff any two snapshots for a change report
diff = manager.diff("v1.0", "v2.0")
print(f"Added: {len(diff.added_nodes)} | Removed: {len(diff.removed_nodes)} | Modified: {len(diff.modified_nodes)}")
for change in diff.changes:
print(f" [{change.type}] {change.element}: {change.description}")
Use verify_checksum() before any compliance export to confirm graph integrity:
from semantica.change_management import verify_checksum
is_valid = verify_checksum(kg, expected_checksum=entry.checksum)
if not is_valid:
raise RuntimeError("Graph has been modified since the snapshot was taken")