class SchemaDiff:
"""Compute a typed structural diff between two schemas."""
def __init__(self, schema_old: Schema, schema_new: Schema):
self.schema_old = schema_old
self.schema_new = schema_new
self.schema_old.finish_init()
self.schema_new.finish_init()
self._result: SchemaDiffResult | None = None
def compare(self) -> SchemaDiffResult:
"""Build a full diff result including operations/conflicts/warnings."""
operations: list[MigrationOperation] = []
conflicts: list[SchemaConflict] = []
warnings: list[str] = []
operations.extend(self._diff_vertices(conflicts))
operations.extend(self._diff_edges(conflicts))
operations.extend(self._diff_database_features())
self._result = SchemaDiffResult(
operations=operations, conflicts=conflicts, warnings=warnings
)
return self._result
def operations(self) -> list[MigrationOperation]:
"""Return diff operations, calculating if needed."""
if self._result is None:
self.compare()
if self._result is None:
return []
return self._result.operations
def is_backward_compatible(self) -> bool:
"""True when schema_new is additive compared to schema_old."""
return is_backward_compatible_operations(self.operations())
def risk_assessment(self) -> dict[str, str]:
"""Map operation keys to risk labels."""
risk_map: dict[str, str] = {}
for op in self.operations():
risk_map[f"{op.op_type}:{op.target}"] = op.risk.value
return risk_map
def validate_union_safety(self) -> list[SchemaConflict]:
"""Return conflicts from latest compare call."""
if self._result is None:
self.compare()
if self._result is None:
return []
return self._result.conflicts
def _diff_vertices(
self, conflicts: list[SchemaConflict]
) -> list[MigrationOperation]:
old_vertices = {
vertex.name: vertex
for vertex in self.schema_old.core_schema.vertex_config.vertices
}
new_vertices = {
vertex.name: vertex
for vertex in self.schema_new.core_schema.vertex_config.vertices
}
old_names = set(old_vertices)
new_names = set(new_vertices)
operations: list[MigrationOperation] = []
for name in sorted(new_names - old_names):
operations.append(
self._op(
OperationType.ADD_VERTEX,
f"vertex:{name}",
None,
new_vertices[name].to_dict(),
)
)
for name in sorted(old_names - new_names):
operations.append(
self._op(
OperationType.REMOVE_VERTEX,
f"vertex:{name}",
old_vertices[name].to_dict(),
None,
reversible=False,
)
)
for name in sorted(old_names & new_names):
old_vertex = old_vertices[name]
new_vertex = new_vertices[name]
operations.extend(
self._diff_vertex_identity(name, old_vertex, new_vertex, conflicts)
)
old_fields = _field_map(old_vertex.properties)
new_fields = _field_map(new_vertex.properties)
old_field_names = set(old_fields)
new_field_names = set(new_fields)
for field_name in sorted(new_field_names - old_field_names):
operations.append(
self._op(
OperationType.ADD_VERTEX_FIELD,
f"vertex:{name}:field:{field_name}",
None,
{"name": field_name, "type": new_fields[field_name]},
)
)
for field_name in sorted(old_field_names - new_field_names):
operations.append(
self._op(
OperationType.REMOVE_VERTEX_FIELD,
f"vertex:{name}:field:{field_name}",
{"name": field_name, "type": old_fields[field_name]},
None,
reversible=False,
)
)
for field_name in sorted(old_field_names & new_field_names):
if old_fields[field_name] != new_fields[field_name]:
operations.append(
self._op(
OperationType.CHANGE_VERTEX_FIELD_TYPE,
f"vertex:{name}:field:{field_name}:type",
old_fields[field_name],
new_fields[field_name],
reversible=False,
)
)
return operations
@staticmethod
def _identity_state(vertex: Vertex) -> dict[str, Any]:
"""Everything that decides how a vertex is keyed at write time.
Comparing only ``identity`` misses a mode change: a vertex moving from a
natural key to a hash keeps ``identity == ["id"]`` on both sides while its
write-time key semantics change completely.
"""
return {
"mode": vertex.identity_mode,
"identity": list(vertex.identity),
"hash_identity_properties": list(vertex.hash_identity_properties),
# A funnel resolves to mode ``hash`` with no flat properties, so the
# branches themselves are the only thing that distinguishes two
# funnel policies — or a funnel from a flat hash.
"identity_funnel": (
vertex.identity_funnel.to_minimal_canonical_dict()
if vertex.identity_funnel is not None
else None
),
}
@staticmethod
def _secondary_identity_state(vertex: Vertex) -> list[dict[str, Any]]:
return sorted(
(
{"name": entry.name, "fields": sorted(entry.fields)}
for entry in vertex.secondary_identities
),
key=lambda entry: (entry["name"] or "", tuple(entry["fields"])),
)
@staticmethod
def _requires_rekey(old_state: dict[str, Any], new_state: dict[str, Any]) -> bool:
"""Whether stored vertex keys become invalid under the new identity.
A mode change always invalidates them — the key is computed differently.
A natural-to-natural change only invalidates them when the new key is not
derivable from the old one; widening a composite key (adding a field) leaves
every existing key still addressable.
"""
if old_state["mode"] != new_state["mode"]:
return True
if new_state["mode"] == "hash":
# Covers flat<->funnel and funnel<->funnel: any change to the digest
# inputs, the branch order or the branch ids yields different keys.
return (
old_state["hash_identity_properties"]
!= new_state["hash_identity_properties"]
or old_state["identity_funnel"] != new_state["identity_funnel"]
)
if new_state["mode"] != "natural":
return False
return not set(old_state["identity"]).issubset(set(new_state["identity"]))
def _diff_vertex_identity(
self,
name: str,
old_vertex: Vertex,
new_vertex: Vertex,
conflicts: list[SchemaConflict],
) -> list[MigrationOperation]:
"""Identity, identity mode, and secondary identities for one vertex."""
operations: list[MigrationOperation] = []
old_state = self._identity_state(old_vertex)
new_state = self._identity_state(new_vertex)
if old_state != new_state:
operations.append(
self._op(
OperationType.CHANGE_VERTEX_IDENTITY,
f"vertex:{name}:identity",
old_state,
new_state,
reversible=False,
)
)
conflicts.append(
SchemaConflict(
key=f"vertex:{name}:identity",
message="Vertex identity changed; requires explicit rekey strategy.",
risk=classify_operation(OperationType.CHANGE_VERTEX_IDENTITY),
)
)
if self._requires_rekey(old_state, new_state):
operations.append(
self._op(
OperationType.REKEY_VERTEX,
f"vertex:{name}:rekey",
old_state,
new_state,
reversible=False,
)
)
conflicts.append(
SchemaConflict(
key=f"vertex:{name}:rekey",
message=(
"Stored vertex keys are no longer derivable from the new "
"identity; existing vertices must be re-keyed."
),
risk=classify_operation(OperationType.REKEY_VERTEX),
)
)
old_secondary = self._secondary_identity_state(old_vertex)
new_secondary = self._secondary_identity_state(new_vertex)
if old_secondary != new_secondary:
operations.append(
self._op(
OperationType.CHANGE_SECONDARY_IDENTITY,
f"vertex:{name}:secondary_identities",
old_secondary,
new_secondary,
)
)
return operations
def _diff_edges(self, conflicts: list[SchemaConflict]) -> list[MigrationOperation]:
old_edges = {
edge.edge_id: edge for edge in self.schema_old.core_schema.edge_config.edges
}
new_edges = {
edge.edge_id: edge for edge in self.schema_new.core_schema.edge_config.edges
}
old_ids = set(old_edges)
new_ids = set(new_edges)
operations: list[MigrationOperation] = []
for edge_id in sorted(new_ids - old_ids):
edge = new_edges[edge_id]
operations.append(
self._op(
OperationType.ADD_EDGE, f"edge:{edge_id}", None, edge.to_dict()
)
)
for edge_id in sorted(old_ids - new_ids):
edge = old_edges[edge_id]
operations.append(
self._op(
OperationType.REMOVE_EDGE,
f"edge:{edge_id}",
edge.to_dict(),
None,
reversible=False,
)
)
for edge_id in sorted(old_ids & new_ids):
old_edge = old_edges[edge_id]
new_edge = new_edges[edge_id]
if old_edge.identities != new_edge.identities:
operations.append(
self._op(
OperationType.CHANGE_EDGE_IDENTITY,
f"edge:{edge_id}:identity",
old_edge.identities,
new_edge.identities,
reversible=False,
)
)
conflicts.append(
SchemaConflict(
key=f"edge:{edge_id}:identity",
message="Edge identity changed; may impact deduplication semantics.",
risk=classify_operation(OperationType.CHANGE_EDGE_IDENTITY),
)
)
old_direct = _field_map(old_edge.properties)
new_direct = _field_map(new_edge.properties)
old_names = set(old_direct)
new_names = set(new_direct)
for field_name in sorted(new_names - old_names):
operations.append(
self._op(
OperationType.ADD_EDGE_FIELD,
f"edge:{edge_id}:field:{field_name}",
None,
{"name": field_name, "type": new_direct[field_name]},
)
)
for field_name in sorted(old_names - new_names):
operations.append(
self._op(
OperationType.REMOVE_EDGE_FIELD,
f"edge:{edge_id}:field:{field_name}",
{"name": field_name, "type": old_direct[field_name]},
None,
reversible=False,
)
)
for field_name in sorted(old_names & new_names):
if old_direct[field_name] != new_direct[field_name]:
operations.append(
self._op(
OperationType.CHANGE_EDGE_FIELD_TYPE,
f"edge:{edge_id}:field:{field_name}:type",
old_direct[field_name],
new_direct[field_name],
reversible=False,
)
)
return operations
def _diff_database_features(self) -> list[MigrationOperation]:
operations: list[MigrationOperation] = []
all_vertices = (
self.schema_old.core_schema.vertex_config.vertex_set
| self.schema_new.core_schema.vertex_config.vertex_set
)
for vertex_name in sorted(all_vertices):
old_ix = (
_vertex_index_tuples(self.schema_old, vertex_name)
if vertex_name in self.schema_old.core_schema.vertex_config.vertex_set
else set()
)
new_ix = (
_vertex_index_tuples(self.schema_new, vertex_name)
if vertex_name in self.schema_new.core_schema.vertex_config.vertex_set
else set()
)
for ix in sorted(new_ix - old_ix):
operations.append(
self._op(
OperationType.ADD_VERTEX_INDEX,
f"vertex:{vertex_name}:index:{ix}",
None,
ix,
)
)
for ix in sorted(old_ix - new_ix):
operations.append(
self._op(
OperationType.REMOVE_VERTEX_INDEX,
f"vertex:{vertex_name}:index:{ix}",
ix,
None,
)
)
old_edges = {
edge.edge_id: edge for edge in self.schema_old.core_schema.edge_config.edges
}
new_edges = {
edge.edge_id: edge for edge in self.schema_new.core_schema.edge_config.edges
}
all_edge_ids = set(old_edges) | set(new_edges)
for edge_id in sorted(all_edge_ids):
old_ix = (
_edge_index_tuples(self.schema_old, old_edges[edge_id])
if edge_id in old_edges
else set()
)
new_ix = (
_edge_index_tuples(self.schema_new, new_edges[edge_id])
if edge_id in new_edges
else set()
)
for ix in sorted(new_ix - old_ix):
operations.append(
self._op(
OperationType.ADD_EDGE_INDEX,
f"edge:{edge_id}:index:{ix}",
None,
ix,
)
)
for ix in sorted(old_ix - new_ix):
operations.append(
self._op(
OperationType.REMOVE_EDGE_INDEX,
f"edge:{edge_id}:index:{ix}",
ix,
None,
)
)
return operations
@staticmethod
def _op(
op_type: OperationType,
target: str,
old_value: Any,
new_value: Any,
reversible: bool = True,
) -> MigrationOperation:
return MigrationOperation(
op_type=op_type,
target=target,
old_value=old_value,
new_value=new_value,
risk=classify_operation(op_type),
reversible=reversible,
)