orchestrata_ast.inheritance

Reference › orchestrata_ast › inheritance

This module answers inheritance questions from the inheritance_graph that the analyzer stores in every snapshot (see models). It is imported by the test analyzer's body_matcher and by several scripts, and tests/test_orchestrata_ast_inheritance.py covers it.

InheritanceMap maps a class name to its ancestor names, closed transitively. from_graph builds one from a plain mapping of class to bases, and from_codebase builds one from any object carrying an inheritance_graph, as described by the _HasInheritanceGraph protocol. None is accepted. add records further direct bases later, for example classes found in test files, and re-closes the map; the closure is cycle-safe. is_subclass is true when the child is the parent or has it as an ancestor, ancestors returns the full set (empty for unknown classes), and iter_mro yields ancestors nearest-first by preorder depth-first search over declared base order.

At a glance

Source src/orchestrata_ast/inheritance.py
Lines of code 118
Classes 2
Functions 0
Imports (internal) none
Imported by orchestrata_ast, orchestrata_tools.test_analyzer.body_matcher, scripts.add_traces_from_evidence, scripts.serialize_codebase_ast
Tests that import it tests/test_orchestrata_ast_inheritance.py

Class diagram

Class diagram orchestrata_ast.inheritance

Dependency diagram

Dependency diagram orchestrata_ast.inheritance

API

Classes

_HasInheritanceGraph (protocol)

Bases: Protocol · Line 21

Any codebase model carrying the serializer's graph (duck-typed so the orchestrata_tools mirror model qualifies without a dependency).

Attribute Type Default
inheritance_graph dict[str, list[str]]

InheritanceMap

Line 28

class name -> ancestor class names, queried transitively.

Method Signature Summary
__init__ __init__() -> None
from_graph classmethod from_graph(graph: Mapping[str, Sequence[str]]) -> InheritanceMap Build from a serializer inheritance_graph (qualified names ok).
from_codebase classmethod from_codebase(source_ast: _HasInheritanceGraph | None) -> InheritanceMap Build from any codebase model carrying inheritance_graph.
add add(child: str, bases: Iterable[str]) -> None Record direct bases discovered later (e.g. classes in test files).
_close_transitively _close_transitively() -> None Flatten ancestors-of-ancestors into _ancestors (cycle-safe).
is_subclass is_subclass(child: str, parent: str) -> bool True if child is parent or has it among its ancestors.
ancestors ancestors(cls_name: str) -> frozenset[str] All recorded ancestors of cls_name (empty if unknown).
iter_mro iter_mro(cls_name: str) -> Iterable[str] Ancestors nearest-first: preorder DFS over declared base order.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ast/inheritance.py

"""Inheritance queries over the AST's own ``inheritance_graph`` — stdlib-only.

The serializer resolves each class's bases into
``CodebaseAST.inheritance_graph`` (qualified name -> direct resolved
bases). This module is the ONE place that turns that raw direct-edge dict
into answers: transitive ``is_subclass``, ``ancestors``, and
deterministic nearest-first ``iter_mro``. Consumers must not hand-roll
inheritance walks — enhance this module instead (user ruling 2026-07-10).

``add()`` exists because some class defs are discovered outside any
snapshot (e.g. classes defined inside test files, parsed live by
``orchestrata_tools``' BodyMatcher).
"""

from __future__ import annotations

from collections.abc import Iterable, Mapping, Sequence
from typing import Protocol


class _HasInheritanceGraph(Protocol):
    """Any codebase model carrying the serializer's graph (duck-typed so the
    orchestrata_tools mirror model qualifies without a dependency)."""

    inheritance_graph: dict[str, list[str]]


class InheritanceMap:
    """class name -> ancestor class names, queried transitively.

    Keys are BARE class names: call-site evidence (``QaVolk.method``) and
    test imports carry bare names, so qualified graph keys are stripped to
    their last segment on ingest.
    """

    def __init__(self) -> None:
        self._ancestors: dict[str, set[str]] = {}
        # direct bases in declaration order — powers the nearest-first
        # DFS of iter_mro (a plain set would randomize precedence)
        self._direct: dict[str, list[str]] = {}

    @classmethod
    def from_graph(cls, graph: Mapping[str, Sequence[str]]) -> InheritanceMap:
        """Build from a serializer ``inheritance_graph`` (qualified names ok)."""
        m = cls()
        for qname, bases in graph.items():
            m.add(
                qname.rsplit(".", 1)[-1],
                [b.rsplit(".", 1)[-1] for b in bases],
            )
        m._close_transitively()
        return m

    @classmethod
    def from_codebase(cls, source_ast: _HasInheritanceGraph | None) -> InheritanceMap:
        """Build from any codebase model carrying ``inheritance_graph``.

        Falls back to walking ``modules[*].classes[*].bases`` (raw
        source-text names) for synthetic codebases whose graph is empty.
        """
        if source_ast is None:
            return cls()
        graph = source_ast.inheritance_graph
        if graph:
            return cls.from_graph(graph)
        m = cls()
        for mod in getattr(source_ast, "modules", []):
            for c in mod.classes:
                m.add(c.name, c.bases)
        m._close_transitively()
        return m

    def add(self, child: str, bases: Iterable[str]) -> None:
        """Record direct bases discovered later (e.g. classes in test files)."""
        bases = [b for b in bases if b]
        if bases:
            self._ancestors.setdefault(child, set()).update(bases)
            direct = self._direct.setdefault(child, [])
            direct.extend(b for b in bases if b not in direct)

    def _close_transitively(self) -> None:
        """Flatten ancestors-of-ancestors into ``_ancestors`` (cycle-safe)."""
        for child in list(self._direct):
            to_process = list(self._direct[child])
            while to_process:
                parent = to_process.pop()
                for b in self._direct.get(parent, []):
                    if b not in self._ancestors.setdefault(child, set()):
                        self._ancestors[child].add(b)
                        to_process.append(b)

    def is_subclass(self, child: str, parent: str) -> bool:
        """True if *child* is *parent* or has it among its ancestors."""
        if child == parent:
            return True
        return parent in self._ancestors.get(child, set())

    def ancestors(self, cls_name: str) -> frozenset[str]:
        """All recorded ancestors of *cls_name* (empty if unknown)."""
        return frozenset(self._ancestors.get(cls_name, set()))

    def iter_mro(self, cls_name: str) -> Iterable[str]:
        """Ancestors nearest-first: preorder DFS over declared base order.

        Deterministic (declaration order, not set order) and cycle-safe.
        Yields ancestor names only, not *cls_name* itself.
        """
        seen: set[str] = set()

        def _walk(name: str) -> Iterable[str]:
            for base in self._direct.get(name, []):
                if base in seen:
                    continue
                seen.add(base)
                yield base
                yield from _walk(base)

        yield from _walk(cls_name)