Schema Evolution¶

Check schema compatibility and compute diffs between versions.

Overview¶

from pycharter.schema_evolution import check_compatibility, compute_diff, CompatibilityMode

result = check_compatibility(old_schema, new_schema, mode=CompatibilityMode.BACKWARD)

API Reference¶

check_compatibility ¶

check_compatibility(
    old_schema: dict[str, Any],
    new_schema: dict[str, Any],
    mode: str | CompatibilityMode = BACKWARD,
) -> CompatibilityResult

Check if two schemas are compatible according to the specified mode.

Compatibility Modes

BACKWARD: New schema can read data produced by old schema (consumers can be upgraded before producers)
FORWARD: Old schema can read data produced by new schema (producers can be upgraded before consumers)
FULL: Both backward and forward compatible

Parameters:

Name	Type	Description	Default
`old_schema`	`dict[str, Any]`	The existing/original schema	required
`new_schema`	`dict[str, Any]`	The new schema to check	required
`mode`	`str \| CompatibilityMode`	Compatibility mode to check against	`BACKWARD`

Returns:

Type	Description
`CompatibilityResult`	CompatibilityResult with compatibility status and details

Example

old = {"type": "object", "properties": {"name": {"type": "string"}}} new = { ... "type": "object", ... "properties": { ... "name": {"type": "string"}, ... "age": {"type": "integer"} # Added optional field ... } ... } result = check_compatibility(old, new, mode="backward") result.compatible True

Source code in src/pycharter/schema_evolution/compatibility.py

def check_compatibility(
    old_schema: dict[str, Any],
    new_schema: dict[str, Any],
    mode: str | CompatibilityMode = CompatibilityMode.BACKWARD,
) -> CompatibilityResult:
    """
    Check if two schemas are compatible according to the specified mode.

    Compatibility Modes:
        - BACKWARD: New schema can read data produced by old schema
                   (consumers can be upgraded before producers)
        - FORWARD: Old schema can read data produced by new schema
                  (producers can be upgraded before consumers)
        - FULL: Both backward and forward compatible

    Args:
        old_schema: The existing/original schema
        new_schema: The new schema to check
        mode: Compatibility mode to check against

    Returns:
        CompatibilityResult with compatibility status and details

    Example:
        >>> old = {"type": "object", "properties": {"name": {"type": "string"}}}
        >>> new = {
        ...     "type": "object",
        ...     "properties": {
        ...         "name": {"type": "string"},
        ...         "age": {"type": "integer"}  # Added optional field
        ...     }
        ... }
        >>> result = check_compatibility(old, new, mode="backward")
        >>> result.compatible
        True
    """
    # Normalize mode
    if isinstance(mode, str):
        mode = CompatibilityMode(mode.lower())

    # Compute diff
    diff = compute_diff(old_schema, new_schema)

    # Check compatibility based on mode
    if mode == CompatibilityMode.BACKWARD:
        return _check_backward_compatibility(diff, mode)
    elif mode == CompatibilityMode.FORWARD:
        return _check_forward_compatibility(diff, mode)
    else:  # FULL
        return _check_full_compatibility(diff, mode)

compute_diff ¶

compute_diff(
    old_schema: dict[str, Any],
    new_schema: dict[str, Any],
    path: str = "",
) -> SchemaDiff

Compute detailed diff between two JSON Schema versions.

Parameters:

Name	Type	Description	Default
`old_schema`	`dict[str, Any]`	Original schema	required
`new_schema`	`dict[str, Any]`	New schema	required
`path`	`str`	Current path (for nested comparison)	`''`

Returns:

Type	Description
`SchemaDiff`	SchemaDiff with all detected changes

Source code in src/pycharter/schema_evolution/diff.py

def compute_diff(
    old_schema: dict[str, Any],
    new_schema: dict[str, Any],
    path: str = "",
) -> SchemaDiff:
    """
    Compute detailed diff between two JSON Schema versions.

    Args:
        old_schema: Original schema
        new_schema: New schema
        path: Current path (for nested comparison)

    Returns:
        SchemaDiff with all detected changes
    """
    changes: list[SchemaChange] = []

    # Compare top-level attributes
    changes.extend(_compare_attributes(old_schema, new_schema, path))

    # Compare properties
    if "properties" in old_schema or "properties" in new_schema:
        changes.extend(
            _compare_properties(
                old_schema.get("properties", {}),
                new_schema.get("properties", {}),
                f"{path}.properties" if path else "properties",
            )
        )

    # Compare required fields
    changes.extend(
        _compare_required(
            old_schema.get("required", []),
            new_schema.get("required", []),
            f"{path}.required" if path else "required",
        )
    )

    # Compare items (for arrays)
    if "items" in old_schema or "items" in new_schema:
        changes.extend(
            _compare_items(
                old_schema.get("items"),
                new_schema.get("items"),
                f"{path}.items" if path else "items",
            )
        )

    return SchemaDiff(changes=changes)

CompatibilityMode¶

Mode	Description
`BACKWARD`	New schema can read old data
`FORWARD`	Old schema can read new data
`FULL`	Both backward and forward
`NONE`	No compatibility check

CompatibilityResult¶

Attribute	Type	Description
`is_compatible`	bool	Compatibility status
`issues`	List[str]	List of issues found

SchemaDiff¶

Attribute	Type	Description
`added_fields`	List[str]	New fields
`removed_fields`	List[str]	Removed fields
`modified_fields`	Dict	Changed fields

Examples¶

from pycharter.schema_evolution import check_compatibility, compute_diff, CompatibilityMode

# Check compatibility
result = check_compatibility(
    old_schema=schema_v1,
    new_schema=schema_v2,
    mode=CompatibilityMode.BACKWARD
)

if not result.is_compatible:
    print(f"Issues: {result.issues}")

# Compute diff
diff = compute_diff(schema_v1, schema_v2)
print(f"Added: {diff.added_fields}")
print(f"Removed: {diff.removed_fields}")