orchestrata_ultimate.refactor.codemod.analysis.breaking_change

Reference › orchestrata_ultimate › refactor › codemod › analysis › breaking_change

BreakingChangeDetector finds which edits in a ChangeSet (see report) change the public signature of a function, and fans each one out to the modules that import it. analyze(target, changes=...) returns a list of BreakingChange records. For each changed file it compares the function signatures before and after (_signatures, _signature_of): the parameters with their kinds and defaults, the return annotation and whether the function is async. Each signature difference becomes a CodeChange that is passed to CodeIntelligenceGraph.breaking_changes in _cig. The consumers it reports are the importing modules, and _import_lines supplies the line of the importing statement. It is read-only and imported by the analysis facade.

At a glance

Class diagram

Dependency diagram

API

Classes

BreakingChangeDetector

Decorators: traces(spec='110', story='US-110-003', reqs=['REQ-110-031', 'REQ-110-032']) · Line 34

Detect API-breaking signature changes and the importers they reach.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/refactor/codemod/analysis/breaking_change.py

"""BreakingChangeDetector — signature diffs of a ChangeSet, fanned out to importers.

GAP-5: ``CodeIntelligenceGraph.breaking_changes`` performs no signature diff;
it lists every dependent of ``change.entity_id`` carrying the change
description. This adapter computes the diff itself, from each
``ChangeEntry``'s original and mutated ``ast`` trees (parameters and return
annotation), and passes only the changed signatures on to the CIG as
``CodeChange`` records.

The entity is the changed module's node: its dependents are the modules that
import it (module granularity, the shape SPEC-110 section 7 uses). A
function that is removed counts as a changed signature.
"""

from __future__ import annotations

import ast
from dataclasses import replace
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod._types import BreakingChange, CodeChange

if TYPE_CHECKING:
    from orchestrata_ultimate.refactor.codemod._cig import CodeIntelligenceGraph
    from orchestrata_ultimate.refactor.codemod.report import ChangeEntry, ChangeSet

__all__ = ["BreakingChangeDetector"]

_FuncDef = ast.FunctionDef | ast.AsyncFunctionDef


@traces(spec="110", story="US-110-003", reqs=["REQ-110-031", "REQ-110-032"])
class BreakingChangeDetector:
    """Detect API-breaking signature changes and the importers they reach."""

    name = "breaking_changes"

    @traces(spec="110", story="US-110-003", reqs=["REQ-110-031", "REQ-110-032"])
    def __init__(self, cig: CodeIntelligenceGraph) -> None:
        self._cig = cig

    @traces(spec="110", story="US-110-003", reqs=["REQ-110-031", "REQ-110-032"])
    def analyze(
        self, target: str | None = None, *, changes: ChangeSet
    ) -> list[BreakingChange]:
        """One BreakingChange per changed signature per importer of its module.

        Args:
            target: Optional module path; only entries for that path count.
            changes: The ChangeSet whose entries are diffed.
        """
        found: list[BreakingChange] = []
        lines: dict[tuple[str, str], int] | None = None
        for entry in changes:
            if target is not None and entry.path != target:
                continue
            for change in self._signature_changes(entry):
                if lines is None:
                    lines = self._import_lines()
                for breaking in self._cig.breaking_changes(change):
                    key = (breaking.consumer_path, breaking.change_path)
                    found.append(
                        replace(
                            breaking,
                            consumer_line=lines.get(key, breaking.consumer_line),
                        )
                    )
        return found

    @traces(spec="110", story="US-110-003", reqs=["REQ-110-031"])
    def _import_lines(self) -> dict[tuple[str, str], int]:
        """First import line of each (importer path, imported module path) pair.

        The CIG reports a consumer at its module node (line 1); the Import
        nodes carry the importing statement's real line and the resolved
        module path (set by the builder's second pass).
        """
        lines: dict[tuple[str, str], int] = {}
        for node in self._cig.nodes.values():
            resolved = node.properties.get("resolved_module")
            if node.node_type != "Import" or resolved is None:
                continue
            key = (node.path, resolved)
            lines[key] = min(node.line, lines.get(key, node.line))
        return lines

    @traces(spec="110", story="US-110-003", reqs=["REQ-110-031"])
    def _signature_changes(self, entry: ChangeEntry) -> list[CodeChange]:
        before = _signatures(entry.original_tree.tree)
        after = _signatures(entry.mutated_tree.tree)
        entity_id = self._cig.key_to_id.get(f"module:{entry.path}")
        changes: list[CodeChange] = []
        for name, old in before.items():
            new = after.get(name)
            if new is not None and self._signature_of(new) == self._signature_of(old):
                continue
            verb = "removed" if new is None else "signature changed"
            changes.append(
                CodeChange(
                    path=entry.path,
                    pass_name=entry.pass_name,
                    description=f"{name}: {verb}",
                    entity_id=entity_id,
                )
            )
        return changes

    @traces(spec="110", story="US-110-003", reqs=["REQ-110-031"])
    def _signature_of(self, node: _FuncDef) -> tuple[str, ...]:
        """``async`` marker, parameters (kind, name, annotation, has-default), return type.

        ``def`` to ``async def`` changes what a caller receives (a coroutine),
        so it counts as a changed signature.
        """
        args = node.args
        parts: list[str] = ["async"] if isinstance(node, ast.AsyncFunctionDef) else []
        positional = [*args.posonlyargs, *args.args]
        default_from = len(positional) - len(args.defaults)
        for index, arg in enumerate(positional):
            kind = "posonly" if index < len(args.posonlyargs) else "arg"
            parts.append(_param(kind, arg, index >= default_from))
        if args.vararg is not None:
            parts.append(_param("vararg", args.vararg, False))
        for arg, default in zip(args.kwonlyargs, args.kw_defaults, strict=True):
            parts.append(_param("kwonly", arg, default is not None))
        if args.kwarg is not None:
            parts.append(_param("kwarg", args.kwarg, False))
        parts.append(f"->{ast.unparse(node.returns) if node.returns else ''}")
        return tuple(parts)


def _param(kind: str, arg: ast.arg, has_default: bool) -> str:
    annotation = ast.unparse(arg.annotation) if arg.annotation else ""
    return f"{kind}:{arg.arg}:{annotation}{'=' if has_default else ''}"


def _signatures(tree: ast.Module) -> dict[str, _FuncDef]:
    """Module-level functions and class methods by dotted in-module name."""
    found: dict[str, _FuncDef] = {}

    def visit(body: list[ast.stmt], prefix: str) -> None:
        for stmt in body:
            if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)):
                found[f"{prefix}{stmt.name}"] = stmt
            elif isinstance(stmt, ast.ClassDef):
                visit(stmt.body, f"{prefix}{stmt.name}.")

    visit(tree.body, "")
    return found