15 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.
Exported Classes
| Class | Role |
|---|---|
TemporalVersionManager |
Snapshot, diff, rollback, and per-node mutation history for KGs |
OntologyVersionManager |
Schema versioning with structural diff support |
InMemoryVersionStorage |
Fast in-memory storage for dev and testing — no persistence |
SQLiteVersionStorage |
Production storage — persists to a local SQLite file |
compute_checksum() |
Returns SHA-256 fingerprint of any dict (graph snapshot, ontology snapshot) |
verify_checksum() |
Detects tampering by recomputing and comparing the stored checksum inside a snapshot dict |
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 checksums on every snapshot to detect any unauthorised modification. Internal metadata validated on every snapshot: ISO 8601 timestamp, email author, and description (max 500 chars). 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")
```
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 |
List and Retrieve
# List all versions — returns List[Dict] with label, author, timestamp, checksum, entity_count
versions = manager.list_versions()
for v in versions:
print(v["label"], "|", v["author"], "|", v["timestamp"], "|", v["checksum"][:8], "...")
# Retrieve a specific version (returns full snapshot dict)
snapshot = manager.get_version("v1.0")
TemporalVersionManager Methods
| Method | Returns | Description |
|---|---|---|
create_snapshot(graph, version_label, author, description) |
Dict[str, Any] |
Create a version snapshot; returns the full snapshot dict including checksum |
get_version(label) |
Optional[Dict[str, Any]] |
Retrieve a snapshot dict for a specific version label |
list_versions() |
List[Dict[str, Any]] |
List all version metadata dicts |
diff(version_a, version_b) |
Dict[str, Any] |
Compare two snapshots; alias for compare_versions |
compare_versions(v1, v2) |
Dict[str, Any] |
Detailed entity/relationship diff between two snapshots |
restore_snapshot(graph, target_version, require_confirmation=True) |
bool |
Restore a live graph to a previous version; raises by default unless require_confirmation=False |
get_node_history(node_id) |
List[Dict[str, Any]] |
Return chronological mutation history for a specific node |
tag_version(version_label, tag_name) |
None |
Create a named tag pointing to a version label |
list_tags() |
Dict[str, str] |
Return mapping of tag name → version label |
prune_versions(keep_last_n) |
Dict[str, Any] |
Delete old snapshots, keeping the most recent N |
verify_checksum(snapshot) |
bool |
Verify snapshot integrity against its stored checksum |
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")
summary = diff["summary"]
print("Entities added: ", summary["entities_added"])
print("Entities removed: ", summary["entities_removed"])
print("Entities modified: ", summary["entities_modified"])
print("Relationships added: ", summary["relationships_added"])
print("Relationships removed:", summary["relationships_removed"])
# Inspect individual modified entities
for item in diff["entities_modified"]:
print("Modified:", item["id"])
for field, change in item["changes"].items():
print(" %s: %s -> %s" % (field, change["from"], change["to"]))
# diff() / compare_versions() returns a plain dict:
{
"version1": str, # first version label
"version2": str, # second version label
"summary": {
"entities_added": int,
"entities_removed": int,
"entities_modified": int,
"relationships_added": int,
"relationships_removed": int,
"relationships_modified": int,
# also present as nodes_*/edges_* aliases
},
"entities_added": List[Dict], # full entity dicts
"entities_removed": List[Dict],
"entities_modified": List[Dict], # {id, before, after, changes}
"relationships_added": List[Dict],
"relationships_removed": List[Dict],
"relationships_modified": List[Dict], # {key, before, after, changes}
# node_*/edge_* aliases point to the same lists
}
OntologyVersionManager
Version control for ontologies — save, diff, and track schema changes:
from semantica.change_management import OntologyVersionManager
manager = OntologyVersionManager()
# Save a version
snapshot = manager.create_snapshot(
ontology_data=ontology,
version_label="1.2.0",
author="ontology-team@example.com",
description="Added FHIR alignment mappings"
)
# Diff two ontology versions — returns a plain dict
diff = manager.compare_versions("1.1.0", "1.2.0")
print("Classes added: ", diff["classes_added"])
print("Classes removed: ", diff["classes_removed"])
print("Properties added: ", diff["properties_added"])
VersionStorage Backends
```python from semantica.change_management import SQLiteVersionStorage, TemporalVersionManager# Pass path directly to the manager (recommended)
manager = TemporalVersionManager(storage_path="versions.db")
```
Persists all version history to disk. Survives process restarts. Recommended for any environment where you need to retain the audit trail.
# Default (no storage_path) uses in-memory storage automatically
manager = TemporalVersionManager()
```
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 a checksum for any dict
checksum = compute_checksum({"nodes": [], "edges": []})
# verify_checksum takes the snapshot dict directly.
# It reads the "checksum" key from the snapshot and recomputes to compare.
snapshot = manager.get_version("v1.0")
is_valid = verify_checksum(snapshot)
if not is_valid:
raise RuntimeError("Snapshot has been tampered with")
ChangeLogEntry
ChangeLogEntry is the internal metadata object created inside create_snapshot. It validates the author (must be a valid email address) and description (non-empty, max 500 characters) before the snapshot is stored.
from semantica.change_management.change_log import ChangeLogEntry
# Create using current timestamp
entry = ChangeLogEntry.create_now(
author="user@example.com", # must be a valid email
description="Initial snapshot" # max 500 chars, non-empty
)
print(entry.timestamp) # ISO 8601 timestamp
print(entry.author) # "user@example.com"
print(entry.description) # "Initial snapshot"
@dataclass
class ChangeLogEntry:
timestamp: str # ISO 8601 timestamp
author: str # valid email address (validated on init)
description: str # change description, max 500 chars
change_id: Optional[str] # optional unique identifier
related_changes: List[str] # optional list of related change IDs
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 v in manager.list_versions():
print(v["timestamp"], "|", v["author"], "|", v["label"], "|", v["description"])
# Diff any two snapshots for a change report
diff = manager.diff("v1.0", "v2.0")
s = diff["summary"]
print("Added: %d | Removed: %d | Modified: %d" % (
s["entities_added"], s["entities_removed"], s["entities_modified"]))
Use verify_checksum() before any compliance export to confirm snapshot integrity:
from semantica.change_management import verify_checksum
snapshot = manager.get_version("v1.0")
is_valid = verify_checksum(snapshot)
if not is_valid:
raise RuntimeError("Snapshot has been modified since it was recorded")
Per-node mutation history is available for HIPAA subject-access and SOX audit workflows:
# Get full mutation history for a specific node
history = manager.get_node_history("patient_001")
for record in history:
print(record["timestamp"], record["operation"], record["version_label"])