feat: add breaking change detection

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- `datacontract breaking` command and `POST /breaking` endpoint for breaking change detection (#1016)
 - `datacontract test` checks the ODCS array options `minItems`, `maxItems` and `uniqueItems` (#1514)
 - `datacontract export odcs` defaults `status` to `draft` when the source DCS contract has no `info.status`
 

README.md

@@ -141,6 +141,9 @@ $ datacontract lint odcs.yaml
 # show a changelog between two data contracts
 $ datacontract changelog v1.odcs.yaml v2.odcs.yaml
 
+# fail when a contract change is backward-incompatible
+$ datacontract breaking v1.odcs.yaml v2.odcs.yaml
+
 # execute schema and quality checks (define credentials as environment variables)
 $ datacontract test odcs.yaml
 

datacontract/api.py

@@ -15,6 +15,7 @@
 
 from datacontract.config import Config, known_env_names
 from datacontract.data_contract import DataContract, ExportFormat
+from datacontract.model.breaking import BreakingChangeEntry
 from datacontract.model.changelog import ChangelogEntry
 from datacontract.model.exceptions import DataContractException, DefinitionResolutionError
 from datacontract.model.run import Check, ResultEnum, Run
@@ -323,6 +324,14 @@ def _cli_version() -> str:
                 "url": "https://docs.datacontract.com/commands/changelog",
             },
         },
+        {
+            "name": "breaking",
+            "description": "Compare two versions of a data contract and grade each change by compatibility impact.",
+            "externalDocs": {
+                "description": "Documentation",
+                "url": "https://docs.datacontract.com/commands/breaking",
+            },
+        },
     ],
 )
 
@@ -881,6 +890,21 @@ class ChangelogResponse(BaseModel):
     )
 
 
+class BreakingChangesResponse(BaseModel):
+    """Every change between two versions of a data contract, each graded `info`, `warning` or `error`."""
+
+    summary: list[BreakingChangeEntry] = Field(
+        description="One entry per changed element, graded by the most severe change. A property whose type and requiredness "
+        "both changed appears once.",
+    )
+    entries: list[BreakingChangeEntry] = Field(
+        description="Every single change with its grade and the old and new value.",
+    )
+    is_breaking: bool = Field(
+        description="Whether any detected change is classified as an error-level breaking change.",
+    )
+
+
 @app.post(
     "/changelog",
     tags=["changelog"],
@@ -935,6 +959,54 @@ async def changelog_endpoint(
         os.unlink(v2_path)
 
 
+@app.post(
+    "/breaking",
+    tags=["breaking"],
+    operation_id="breakingChangesDataContracts",
+    summary="Show compatibility impact between two data contracts.",
+    description="""
+Compare two versions of an ODCS data contract and detect backward-incompatible changes.
+
+`POST` a JSON body with `v1` (before) and `v2` (after) as YAML strings. A contract that cannot be
+parsed is answered with `422`.
+    """,
+    response_description="The breaking changes detected between the two data contracts.",
+    responses={
+        **AUTHENTICATION_RESPONSES,
+        422: {
+            "description": "One of the two data contracts is not valid YAML or not a valid data contract.",
+            "model": UnprocessableEntityResponse,
+            "content": {"application/json": {"example": {"detail": "Invalid YAML: while parsing a block mapping"}}},
+        },
+    },
+)
+def breaking_endpoint(
+    body: ChangelogRequest,
+    api_key: Annotated[str | None, Depends(api_key_header)] = None,
+) -> BreakingChangesResponse:
+    check_api_key(api_key)
+
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f1:
+        f1.write(body.v1)
+        v1_path = f1.name
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f2:
+        f2.write(body.v2)
+        v2_path = f2.name
+
+    try:
+        result = DataContract(data_contract_file=v1_path).breaking(DataContract(data_contract_file=v2_path))
+        return BreakingChangesResponse(summary=result.summary, entries=result.entries, is_breaking=result.is_breaking)
+    except yaml.YAMLError as e:
+        raise HTTPException(status_code=422, detail=f"Invalid YAML: {e}")
+    except pydantic.ValidationError as e:
+        raise HTTPException(status_code=422, detail=f"Invalid data contract: {e}")
+    except DataContractException as e:
+        raise HTTPException(status_code=422, detail=f"Data Contract Validation Failure: {e}")
+    finally:
+        os.unlink(v1_path)
+        os.unlink(v2_path)
+
+
 @app.post(
     "/export",
     tags=["export"],

datacontract/breaking/detector.py

@@ -0,0 +1,74 @@
+from collections import defaultdict
+from collections.abc import Iterable
+
+from datacontract.breaking.rules import DEFAULT_RULES, BreakingChangeRule
+from datacontract.model.breaking import BreakingChangeEntry, BreakingChangeLevel, BreakingChangeResult
+from datacontract.model.changelog import ChangelogEntry, ChangelogResult, ChangelogType
+
+_LEVEL_ORDER = {
+    BreakingChangeLevel.INFO: 0,
+    BreakingChangeLevel.WARNING: 1,
+    BreakingChangeLevel.ERROR: 2,
+}
+
+
+class BreakingChangeDetector:
+    """Classify detailed changelog entries using an ordered rule set."""
+
+    def __init__(self, rules: Iterable[BreakingChangeRule] = DEFAULT_RULES):
+        self._rules = tuple(rules)
+        if not self._rules:
+            raise ValueError("At least one breaking-change rule is required")
+
+    def detect(self, changelog: ChangelogResult) -> BreakingChangeResult:
+        entries = [self._classify(entry) for entry in changelog.entries]
+        # Nothing inside a newly added element can break a consumer: it did not exist before.
+        added = {entry.path for entry in entries if entry.change_type == ChangelogType.added}
+        for entry in entries:
+            segments = entry.path.split(".")
+            if any(".".join(segments[:i]) in added for i in range(1, len(segments))):
+                entry.level = BreakingChangeLevel.INFO
+        entries_by_prefix = defaultdict(list)
+        for entry in entries:
+            prefix = ""
+            for segment in entry.path.split("."):
+                prefix = f"{prefix}.{segment}" if prefix else segment
+                entries_by_prefix[prefix].append(entry)
+        summary = [self._summarize(entry, entries_by_prefix.get(entry.path, [])) for entry in changelog.summary]
+        return BreakingChangeResult(v1=changelog.v1, v2=changelog.v2, summary=summary, entries=entries)
+
+    def _classify(self, entry: ChangelogEntry) -> BreakingChangeEntry:
+        for rule in self._rules:
+            evaluation = rule.evaluate(entry)
+            if evaluation is not None:
+                return BreakingChangeEntry(
+                    path=entry.path,
+                    change_type=entry.type,
+                    level=evaluation.level,
+                    message=evaluation.message,
+                    rule_id=evaluation.rule_id,
+                    old_value=entry.old_value,
+                    new_value=entry.new_value,
+                )
+        raise ValueError(f"No breaking-change rule classified {entry.path}")
+
+    @staticmethod
+    def _summarize(summary_entry: ChangelogEntry, entries: list[BreakingChangeEntry]) -> BreakingChangeEntry:
+        if not entries:
+            return BreakingChangeEntry(
+                path=summary_entry.path,
+                change_type=summary_entry.type,
+                level=BreakingChangeLevel.INFO,
+                message=f"Changed contract at {summary_entry.path}",
+                rule_id="summary-fallback",
+            )
+        highest = max(entries, key=lambda entry: _LEVEL_ORDER[entry.level])
+        return BreakingChangeEntry(
+            path=summary_entry.path,
+            change_type=summary_entry.type,
+            level=highest.level,
+            message=f"{highest.level.value.capitalize()} impact at {summary_entry.path}",
+            rule_id=f"summary:{highest.rule_id}",
+            old_value=highest.old_value,
+            new_value=highest.new_value,
+        )

datacontract/breaking/rules.py

@@ -0,0 +1,196 @@
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from datetime import date
+
+from datacontract.model.breaking import BreakingChangeLevel
+from datacontract.model.changelog import ChangelogEntry, ChangelogType
+
+
+@dataclass(frozen=True)
+class RuleEvaluation:
+    rule_id: str
+    level: BreakingChangeLevel
+    message: str
+
+
+class BreakingChangeRule(ABC):
+    rule_id: str
+
+    @abstractmethod
+    def evaluate(self, entry: ChangelogEntry) -> RuleEvaluation | None:
+        """Return a classification when this rule applies to ``entry``."""
+
+
+class SchemaRemovedRule(BreakingChangeRule):
+    rule_id = "schema-removed"
+
+    def evaluate(self, entry: ChangelogEntry) -> RuleEvaluation | None:
+        segments = entry.path.split(".")
+        if entry.type == ChangelogType.removed and len(segments) == 2 and segments[0] == "schema":
+            return RuleEvaluation(self.rule_id, BreakingChangeLevel.ERROR, f"Removed schema {segments[1]}")
+        return None
+
+
+class FieldRemovedRule(BreakingChangeRule):
+    rule_id = "field-removed"
+
+    def evaluate(self, entry: ChangelogEntry) -> RuleEvaluation | None:
+        segments = entry.path.split(".")
+        if entry.type != ChangelogType.removed or not _is_schema_property_path(segments):
+            return None
+        property_name = segments[-1]
+        return RuleEvaluation(self.rule_id, BreakingChangeLevel.ERROR, f"Removed property {property_name}")
+
+
+class RequiredChangedRule(BreakingChangeRule):
+    rule_id = "required-changed"
+
+    def evaluate(self, entry: ChangelogEntry) -> RuleEvaluation | None:
+        if not entry.path.endswith(".required"):
+            return None
+        old = _parse_bool(entry.old_value)
+        new = _parse_bool(entry.new_value)
+        if entry.type == ChangelogType.added and new is True:
+            level = BreakingChangeLevel.ERROR
+        elif entry.type == ChangelogType.removed or new is False:
+            # Dropping the requirement, or writing out the optional default, only loosens the contract.
+            level = BreakingChangeLevel.INFO
+        elif entry.type == ChangelogType.updated and old is False and new is True:
+            level = BreakingChangeLevel.ERROR
+        elif entry.type in (ChangelogType.added, ChangelogType.updated):
+            level = BreakingChangeLevel.WARNING
+        else:
+            return None
+        return RuleEvaluation(self.rule_id, level, _change_message("requiredness", entry))
+
+
+class TypeChangedRule(BreakingChangeRule):
+    rule_id = "type-changed"
+
+    def evaluate(self, entry: ChangelogEntry) -> RuleEvaluation | None:
+        if not entry.path.endswith((".logicalType", ".physicalType")):
+            return None
+        if entry.type == ChangelogType.updated:
+            level = BreakingChangeLevel.ERROR
+        elif entry.type == ChangelogType.added:
+            level = BreakingChangeLevel.INFO
+        elif entry.type == ChangelogType.removed:
+            level = BreakingChangeLevel.WARNING
+        else:
+            return None
+        return RuleEvaluation(self.rule_id, level, _change_message("type", entry))
+
+
+class UniqueConstraintRule(BreakingChangeRule):
+    rule_id = "unique-constraint-changed"
+
+    def evaluate(self, entry: ChangelogEntry) -> RuleEvaluation | None:
+        if not entry.path.endswith(".unique"):
+            return None
+        old = _parse_bool(entry.old_value)
+        new = _parse_bool(entry.new_value)
+        if new is True and old is not True:
+            level = BreakingChangeLevel.ERROR
+        elif old is True and new is not True:
+            level = BreakingChangeLevel.INFO
+        else:
+            level = BreakingChangeLevel.WARNING
+        return RuleEvaluation(self.rule_id, level, _change_message("uniqueness", entry))
+
+
+class KeyConstraintRule(BreakingChangeRule):
+    rule_id = "key-constraint-changed"
+
+    def evaluate(self, entry: ChangelogEntry) -> RuleEvaluation | None:
+        if not entry.path.endswith((".primaryKey", ".primary_key")):
+            return None
+        return RuleEvaluation(self.rule_id, BreakingChangeLevel.WARNING, _change_message("key constraint", entry))
+
+
+class ValidationConstraintRule(BreakingChangeRule):
+    rule_id = "validation-constraint-changed"
+    _suffixes = (
+        ".pattern",
+        ".minLength",
+        ".maxLength",
+        ".minimum",
+        ".maximum",
+        ".exclusiveMinimum",
+        ".exclusiveMaximum",
+    )
+
+    def evaluate(self, entry: ChangelogEntry) -> RuleEvaluation | None:
+        if not entry.path.endswith(self._suffixes):
+            return None
+        old = _parse_number(entry.old_value)
+        new = _parse_number(entry.new_value)
+        if old is None or new is None:
+            level = BreakingChangeLevel.WARNING
+        elif _is_tightening(entry.path, old, new):
+            level = BreakingChangeLevel.ERROR
+        else:
+            level = BreakingChangeLevel.INFO
+        return RuleEvaluation(self.rule_id, level, _change_message("validation constraint", entry))
+
+
+class MetadataFallbackRule(BreakingChangeRule):
+    rule_id = "metadata-or-unknown-change"
+
+    def evaluate(self, entry: ChangelogEntry) -> RuleEvaluation:
+        return RuleEvaluation(self.rule_id, BreakingChangeLevel.INFO, _change_message("contract", entry))
+
+
+def _is_schema_property_path(segments: list[str]) -> bool:
+    return len(segments) >= 3 and segments[0] == "schema" and segments[-2] == "properties"
+
+
+def _parse_bool(value: str | None) -> bool | None:
+    if value is None:
+        return None
+    normalized = value.strip().lower()
+    if normalized in {"true", "1"}:
+        return True
+    if normalized in {"false", "0"}:
+        return False
+    return None
+
+
+def _parse_number(value: str | None) -> float | None:
+    if value is None:
+        return None
+    normalized = value.strip()
+    try:
+        return float(normalized)
+    except ValueError:
+        try:
+            return float(date.fromisoformat(normalized).toordinal())
+        except ValueError:
+            return None
+
+
+def _is_tightening(path: str, old: float, new: float) -> bool:
+    if path.endswith((".minLength", ".minimum", ".exclusiveMinimum")):
+        return new > old
+    if path.endswith((".maxLength", ".maximum", ".exclusiveMaximum")):
+        return new < old
+    return True
+
+
+def _change_message(subject: str, entry: ChangelogEntry) -> str:
+    if entry.type == ChangelogType.added:
+        return f"Added {subject} at {entry.path}"
+    if entry.type == ChangelogType.removed:
+        return f"Removed {subject} at {entry.path}"
+    return f"Changed {subject} at {entry.path} from {entry.old_value!r} to {entry.new_value!r}"
+
+
+DEFAULT_RULES: tuple[BreakingChangeRule, ...] = (
+    RequiredChangedRule(),
+    SchemaRemovedRule(),
+    FieldRemovedRule(),
+    TypeChangedRule(),
+    UniqueConstraintRule(),
+    KeyConstraintRule(),
+    ValidationConstraintRule(),
+    MetadataFallbackRule(),
+)