type_bridge.migration.breaking

breaking

Breaking change analysis for TypeDB schema migrations.

This module provides classification of schema changes to help determine which changes are safe, require warnings, or are breaking changes.

ChangeCategory

Bases: Enum

Classification of schema changes by severity.

SAFE class-attribute instance-attribute

SAFE = 'safe'

Backwards compatible change - no data loss or errors.

WARNING class-attribute instance-attribute

WARNING = 'warning'

May cause issues - review required.

BREAKING class-attribute instance-attribute

BREAKING = 'breaking'

Will cause data loss or errors - requires migration plan.

ClassifiedChange dataclass

ClassifiedChange(description, category, recommendation)

A schema change with its classification and recommendation.

BreakingChangeAnalyzer

Analyzes schema diffs to classify changes by severity.

Classification rules: - SAFE: Adding new types, widening role player types - WARNING: Adding required attributes to existing types - BREAKING: Removing types, narrowing role player types, removing roles

Example

analyzer = BreakingChangeAnalyzer() diff = old_schema.compare(new_schema) changes = analyzer.analyze(diff)

for change in changes: print(f"[{change.category.value}] {change.description}") print(f" Recommendation: {change.recommendation}")

analyze

analyze(diff)

Classify all changes in the schema diff.

Parameters:

Name	Type	Description	Default
diff	SchemaDiff	SchemaDiff from SchemaInfo.compare()	required

Returns:

Type	Description
list[ClassifiedChange]	List of classified changes with recommendations

Source code in type_bridge/migration/breaking.py

def analyze(self, diff: SchemaDiff) -> list[ClassifiedChange]:
    """Classify all changes in the schema diff.

    Args:
        diff: SchemaDiff from SchemaInfo.compare()

    Returns:
        List of classified changes with recommendations
    """
    changes: list[ClassifiedChange] = []

    # Analyze entity changes
    changes.extend(self._analyze_entity_changes(diff))

    # Analyze relation changes
    changes.extend(self._analyze_relation_changes(diff))

    # Analyze attribute changes
    changes.extend(self._analyze_attribute_changes(diff))

    return changes

has_breaking_changes

has_breaking_changes(diff)

Quick check for any breaking changes.

Parameters:

Name	Type	Description	Default
diff	SchemaDiff	SchemaDiff from SchemaInfo.compare()	required

Returns:

Type	Description
bool	True if any breaking changes exist

Source code in type_bridge/migration/breaking.py

def has_breaking_changes(self, diff: SchemaDiff) -> bool:
    """Quick check for any breaking changes.

    Args:
        diff: SchemaDiff from SchemaInfo.compare()

    Returns:
        True if any breaking changes exist
    """
    classified = self.analyze(diff)
    return any(c.category == ChangeCategory.BREAKING for c in classified)

has_warnings

has_warnings(diff)

Quick check for any warning-level changes.

Parameters:

Name	Type	Description	Default
diff	SchemaDiff	SchemaDiff from SchemaInfo.compare()	required

Returns:

Type	Description
bool	True if any warnings exist

Source code in type_bridge/migration/breaking.py

def has_warnings(self, diff: SchemaDiff) -> bool:
    """Quick check for any warning-level changes.

    Args:
        diff: SchemaDiff from SchemaInfo.compare()

    Returns:
        True if any warnings exist
    """
    classified = self.analyze(diff)
    return any(c.category == ChangeCategory.WARNING for c in classified)

get_breaking_changes

get_breaking_changes(diff)

Get only breaking changes from the diff.

Parameters:

Name	Type	Description	Default
diff	SchemaDiff	SchemaDiff from SchemaInfo.compare()	required

Returns:

Type	Description
list[ClassifiedChange]	List of breaking changes only

Source code in type_bridge/migration/breaking.py

def get_breaking_changes(self, diff: SchemaDiff) -> list[ClassifiedChange]:
    """Get only breaking changes from the diff.

    Args:
        diff: SchemaDiff from SchemaInfo.compare()

    Returns:
        List of breaking changes only
    """
    return [c for c in self.analyze(diff) if c.category == ChangeCategory.BREAKING]

summary

summary(diff)

Generate a human-readable summary of classified changes.

Parameters:

Name	Type	Description	Default
diff	SchemaDiff	SchemaDiff from SchemaInfo.compare()	required

Returns:

Type	Description
str	Formatted summary string

Source code in type_bridge/migration/breaking.py

def summary(self, diff: SchemaDiff) -> str:
    """Generate a human-readable summary of classified changes.

    Args:
        diff: SchemaDiff from SchemaInfo.compare()

    Returns:
        Formatted summary string
    """
    classified = self.analyze(diff)

    if not classified:
        return "No schema changes detected."

    lines = ["Schema Change Analysis", "=" * 50]

    # Group by category
    breaking = [c for c in classified if c.category == ChangeCategory.BREAKING]
    warnings = [c for c in classified if c.category == ChangeCategory.WARNING]
    safe = [c for c in classified if c.category == ChangeCategory.SAFE]

    if breaking:
        lines.append(f"\n[BREAKING] ({len(breaking)} changes)")
        for change in breaking:
            lines.append(f"  - {change.description}")
            lines.append(f"    Recommendation: {change.recommendation}")

    if warnings:
        lines.append(f"\n[WARNING] ({len(warnings)} changes)")
        for change in warnings:
            lines.append(f"  - {change.description}")
            lines.append(f"    Recommendation: {change.recommendation}")

    if safe:
        lines.append(f"\n[SAFE] ({len(safe)} changes)")
        for change in safe:
            lines.append(f"  - {change.description}")

    return "\n".join(lines)