graflo.architecture.evolution.naming_graph

One naming graph over the original names of both sides of a merge.

The nodes are the classes and relations each manifest declares, under their own names, never erased. Three declarations add edges:

• the vocabulary (canonical_maps) — a class's default name in the union; several classes a side's vocabulary sends to one name are merged by it, which its allow_merges acknowledges;
• an equivalence — its members are one class, across the two sides;
• renames — a class or relation no group holds, called something else.

A group is a connected component over the merging edges (vocabulary merges, equivalence membership, and the pairs union_right unions), so an equivalence naming one class a vocabulary merges with others takes all of them. A group is named by its equivalences' into (a name in the union, never translated), else by the vocabulary, else by the one spelling its members share. Everything not in a group keeps its rename, else its vocabulary name, else its own.

The merge is valid when every union name is reached by exactly one group or one ungrouped class, every declared name exists, and every group has one name and one identity. :func:build_naming reports every violation as a :class:NamingFinding instead of raising, so a caller sees all of them at once; :func:resolve_naming raises them together as one :class:MergeNamingError.

Per-member maps stay keyed by the members' own names, and :func:~graflo.architecture.evolution.merge.merge_manifests reads each side as handed in, before the relabel. So class-specific identity sources, property maps and the when guards derived from how a resource produces a member survive the merge. A member a vocabulary joins whose producing resources the group's identity does not key is given its own key behind a tag (side:Class): it belongs to the class, but its records never fuse with another member's.

Attributes

NodeKey = tuple[Kind, Side, str] module-attribute

Severity = Literal['refusal', 'incomplete', 'note'] module-attribute

Via = Literal['equivalence', 'vocabulary', 'rename', 'union_right', 'own name'] module-attribute

__all__ = ['MergeNamingError', 'MergeNamingIncompleteError', 'NamingEdge', 'NamingFinding', 'NamingGraph', 'NamingGroup', 'NamingResult', 'apply_repair', 'build_naming', 'naming_table', 'resolve_naming', 'suggest_merge_op'] module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

MergeNamingError

Bases: MergeCanonicalConflictError

Every naming problem of one merge op, raised together.

findings lists them; check and completion are the first one's, and subjects names every node any of them is about.

Source code in graflo/architecture/evolution/naming_graph.py

class MergeNamingError(MergeCanonicalConflictError):
    """Every naming problem of one merge op, raised together.

    ``findings`` lists them; ``check`` and ``completion`` are the first one's,
    and ``subjects`` names every node any of them is about.
    """

    def __init__(self, findings: Sequence[NamingFinding]) -> None:
        blocking = tuple(f for f in findings if f.blocking) or tuple(findings)
        Refusal.__init__(
            self,
            _describe_findings(blocking),
            check=blocking[0].check,
            subjects=tuple(dict.fromkeys(s for f in blocking for s in f.subjects)),
        )
        self.findings = blocking
        self.completion: Completion | None = (
            blocking[0].repairs[0] if blocking[0].repairs else None
        )

Attributes

completion = blocking[0].repairs[0] if blocking[0].repairs else None instance-attribute

findings = blocking instance-attribute

Methods:

__init__(findings)

Source code in graflo/architecture/evolution/naming_graph.py

def __init__(self, findings: Sequence[NamingFinding]) -> None:
    blocking = tuple(f for f in findings if f.blocking) or tuple(findings)
    Refusal.__init__(
        self,
        _describe_findings(blocking),
        check=blocking[0].check,
        subjects=tuple(dict.fromkeys(s for f in blocking for s in f.subjects)),
    )
    self.findings = blocking
    self.completion: Completion | None = (
        blocking[0].repairs[0] if blocking[0].repairs else None
    )

MergeNamingIncompleteError

Bases: MergeNamingError, MergeIncompleteError

Every naming problem is one an addition settles; completion says which.

Source code in graflo/architecture/evolution/naming_graph.py

class MergeNamingIncompleteError(MergeNamingError, MergeIncompleteError):
    """Every naming problem is one an addition settles; ``completion`` says which."""

NamingEdge dataclass

One thing a declaration says about where a name goes.

Source code in graflo/architecture/evolution/naming_graph.py

@dataclass(frozen=True)
class NamingEdge:
    """One thing a declaration says about where a name goes."""

    kind: Literal["member", "vocabulary", "rename", "union_right"]
    category: Kind
    side: Side
    source: str
    target: str | None
    declared_by: str

Attributes

category instance-attribute

declared_by instance-attribute

kind instance-attribute

side instance-attribute

source instance-attribute

target instance-attribute

Methods:

__init__(kind, category, side, source, target, declared_by)

NamingFinding dataclass

One violation of the naming rules, or one acknowledged default.

kind is the rule, subjects the names it is about as :func:~graflo.architecture.evolution.equivalence.subject ids, and repairs the declarations that would settle it, safest first.

Source code in graflo/architecture/evolution/naming_graph.py

@dataclass(frozen=True)
class NamingFinding:
    """One violation of the naming rules, or one acknowledged default.

    ``kind`` is the rule, ``subjects`` the names it is about as
    :func:`~graflo.architecture.evolution.equivalence.subject` ids, and
    ``repairs`` the declarations that would settle it, safest first.
    """

    kind: str
    severity: Severity
    message: str
    subjects: tuple[str, ...] = ()
    repairs: tuple[Completion, ...] = ()
    check: str = ""

    @property
    def blocking(self) -> bool:
        """Whether this finding stops the merge."""
        return self.severity != "note"

Attributes

blocking property

Whether this finding stops the merge.

check = '' class-attribute instance-attribute

kind instance-attribute

message instance-attribute

repairs = () class-attribute instance-attribute

severity instance-attribute

subjects = () class-attribute instance-attribute

Methods:

__init__(kind, severity, message, subjects=(), repairs=(), check='')

NamingGraph dataclass

Every node's union name and provenance, the groups, and the edges that put them there.

Source code in graflo/architecture/evolution/naming_graph.py

@dataclass(frozen=True)
class NamingGraph:
    """Every node's union name and provenance, the groups, and the edges that put them there."""

    targets: Mapping[NodeKey, str | None]
    via: Mapping[NodeKey, Via]
    groups: tuple[NamingGroup, ...]
    edges: tuple[NamingEdge, ...]

    def rows(
        self, kind: Kind = "vertex"
    ) -> list[tuple[str, list[str], list[str], str]]:
        """``(union name, left names, right names, via)`` per union name of *kind*, sorted.

        One row per provenance, so a group joined partly through a vocabulary
        shows the vocabulary-joined members on their own row.
        """
        out: dict[tuple[str, str], tuple[list[str], list[str]]] = {}
        for (node_kind, side, name), target in self.targets.items():
            if node_kind != kind:
                continue
            union = target if target is not None else "?"
            via = self.via.get((node_kind, side, name), "own name")
            left, right = out.setdefault((union, via), ([], []))
            (left if side == "left" else right).append(name)
        order = {"equivalence": 0, "union_right": 1, "vocabulary": 2, "rename": 3}
        return [
            (union, sorted(left), sorted(right), via)
            for (union, via), (left, right) in sorted(
                out.items(), key=lambda item: (item[0][0], order.get(item[0][1], 4))
            )
        ]

Attributes

edges instance-attribute

groups instance-attribute

targets instance-attribute

via instance-attribute

Methods:

__init__(targets, via, groups, edges)

rows(kind='vertex')

(union name, left names, right names, via) per union name of kind, sorted.

One row per provenance, so a group joined partly through a vocabulary shows the vocabulary-joined members on their own row.

Source code in graflo/architecture/evolution/naming_graph.py

def rows(
    self, kind: Kind = "vertex"
) -> list[tuple[str, list[str], list[str], str]]:
    """``(union name, left names, right names, via)`` per union name of *kind*, sorted.

    One row per provenance, so a group joined partly through a vocabulary
    shows the vocabulary-joined members on their own row.
    """
    out: dict[tuple[str, str], tuple[list[str], list[str]]] = {}
    for (node_kind, side, name), target in self.targets.items():
        if node_kind != kind:
            continue
        union = target if target is not None else "?"
        via = self.via.get((node_kind, side, name), "own name")
        left, right = out.setdefault((union, via), ([], []))
        (left if side == "left" else right).append(name)
    order = {"equivalence": 0, "union_right": 1, "vocabulary": 2, "rename": 3}
    return [
        (union, sorted(left), sorted(right), via)
        for (union, via), (left, right) in sorted(
            out.items(), key=lambda item: (item[0][0], order.get(item[0][1], 4))
        )
    ]

NamingGroup dataclass

One group: its union name, and each member with where its name came from.

Source code in graflo/architecture/evolution/naming_graph.py

@dataclass(frozen=True)
class NamingGroup:
    """One group: its union name, and each member with where its name came from."""

    kind: Kind
    name: str | None
    members: tuple[tuple[Side, str, Via], ...]
    equivalences: tuple[int, ...] = ()
    synthesized: bool = False

    def side_members(self, side: Side) -> tuple[str, ...]:
        return tuple(name for s, name, _via in self.members if s == side)

Attributes

equivalences = () class-attribute instance-attribute

kind instance-attribute

members instance-attribute

name instance-attribute

synthesized = False class-attribute instance-attribute

Methods:

__init__(kind, name, members, equivalences=(), synthesized=False)

side_members(side)

Source code in graflo/architecture/evolution/naming_graph.py

def side_members(self, side: Side) -> tuple[str, ...]:
    return tuple(name for s, name, _via in self.members if s == side)

NamingResult dataclass

What a merge would apply, and everything wrong with its declarations.

Source code in graflo/architecture/evolution/naming_graph.py

@dataclass(frozen=True)
class NamingResult:
    """What a merge would apply, and everything wrong with its declarations."""

    resolution: ClusterResolution
    graph: NamingGraph
    findings: tuple[NamingFinding, ...]

    @property
    def blocking(self) -> tuple[NamingFinding, ...]:
        return tuple(f for f in self.findings if f.blocking)

    @property
    def notes(self) -> tuple[NamingFinding, ...]:
        return tuple(f for f in self.findings if not f.blocking)

    def raise_if_blocking(self) -> None:
        """Raise every blocking finding as one :class:`MergeNamingError`."""
        blocking = self.blocking
        if not blocking:
            return
        if all(f.severity == "incomplete" for f in blocking):
            raise MergeNamingIncompleteError(blocking)
        raise MergeNamingError(blocking)

Attributes

blocking property

findings instance-attribute

graph instance-attribute

notes property

resolution instance-attribute

Methods:

__init__(resolution, graph, findings)

raise_if_blocking()

Raise every blocking finding as one :class:MergeNamingError.

Source code in graflo/architecture/evolution/naming_graph.py

def raise_if_blocking(self) -> None:
    """Raise every blocking finding as one :class:`MergeNamingError`."""
    blocking = self.blocking
    if not blocking:
        return
    if all(f.severity == "incomplete" for f in blocking):
        raise MergeNamingIncompleteError(blocking)
    raise MergeNamingError(blocking)

Functions:

apply_repair(op, repair)

op with repair applied, or None when an edit of the op cannot express it.

add_key_source carries an identity entry to fill in by hand, and a repair of a synthesized group has no declaration to replace.

Source code in graflo/architecture/evolution/naming_graph.py

def apply_repair(op: MergeManifestsOp, repair: Completion) -> MergeManifestsOp | None:
    """*op* with *repair* applied, or ``None`` when an edit of the op cannot express it.

    ``add_key_source`` carries an identity entry to fill in by hand, and a
    repair of a synthesized group has no declaration to replace.
    """
    payload = op.to_dict(skip_defaults=True)
    if repair.kind == "declare_equivalences":
        for key, added in (
            ("vertex_equivalences", repair.vertex_equivalences),
            ("relation_equivalences", repair.relation_equivalences),
        ):
            if added:
                payload[key] = [*payload.get(key, []), *(dict(p) for p in added)]
    elif repair.kind in ("set_into", "extend_cluster"):
        if repair.replaces is None:
            return None
        key = (
            "vertex_equivalences"
            if repair.vertex_equivalences
            else "relation_equivalences"
        )
        (replacement,) = repair.vertex_equivalences or repair.relation_equivalences
        declared = list(payload.get(key, []))
        if repair.replaces >= len(declared):
            return None
        declared[repair.replaces] = dict(replacement)
        payload[key] = declared
    elif repair.kind == "rename_away":
        renames = payload.setdefault("renames", {})
        for side, fragment in repair.renames.items():
            side_renames = renames.setdefault(side, {})
            for field_name, mapping in fragment.items():
                side_renames[field_name] = {
                    **side_renames.get(field_name, {}),
                    **mapping,
                }
    else:
        return None
    return MergeManifestsOp.model_validate(payload)

build_naming(op, *, left, right, canonical_maps=())

Resolve op against left and right in one pass, refusing nothing.

Parameters:

Name	Type	Description	Default
op	MergeManifestsOp	The merge op: equivalences, vocabulary, renames, policy.	required
left	GraphManifest	The left manifest, as handed to merge.	required
right	GraphManifest	The right manifest.	required
canonical_maps	Sequence[tuple[Side, CanonicalMap]]	Extra (side, map) vocabulary pairs, folded into op.canonical_maps.	()

Returns:

Type	Description
NamingResult	The resolution merge would apply (groups as clusters over their closed
NamingResult	member sets, one relabel per side), the naming graph, and every
NamingResult	finding. Lowered best-effort when a finding blocks it.

Source code in graflo/architecture/evolution/naming_graph.py

def build_naming(
    op: MergeManifestsOp,
    *,
    left: GraphManifest,
    right: GraphManifest,
    canonical_maps: Sequence[tuple[Side, CanonicalMap]] = (),
) -> NamingResult:
    """Resolve *op* against *left* and *right* in one pass, refusing nothing.

    Args:
        op: The merge op: equivalences, vocabulary, renames, policy.
        left: The left manifest, as handed to merge.
        right: The right manifest.
        canonical_maps: Extra ``(side, map)`` vocabulary pairs, folded into
            ``op.canonical_maps``.

    Returns:
        The resolution merge would apply (groups as clusters over their closed
        member sets, one relabel per side), the naming graph, and every
        finding. Lowered best-effort when a finding blocks it.
    """
    return _Naming(
        op=op, manifests={"left": left, "right": right}, extra=canonical_maps
    ).run()

naming_table(graph, kind='vertex', *, findings=(), quiet=False)

The naming graph as text: one row per union name and provenance.

A row whose union name a blocking finding is about ends in <- conflict. With quiet, a table where every class keeps its own name and nothing conflicts is left out.

Source code in graflo/architecture/evolution/naming_graph.py

def naming_table(
    graph: NamingGraph,
    kind: Kind = "vertex",
    *,
    findings: Sequence[NamingFinding] = (),
    quiet: bool = False,
) -> list[str]:
    """The naming graph as text: one row per union name and provenance.

    A row whose union name a blocking finding is about ends in
    ``<- conflict``. With *quiet*, a table where every class keeps its own
    name and nothing conflicts is left out.
    """
    rows = graph.rows(kind)
    marked = {
        s.removeprefix("merged:")
        for f in findings
        if f.blocking
        for s in f.subjects
        if s.startswith("merged:")
    }
    if not rows or (
        quiet
        and all(via == "own name" for *_r, via in rows)
        and not {union for union, *_r in rows} & marked
    ):
        return []
    header = ("merged", "left", "right", "via")
    cells = [
        (union, ", ".join(left) or "-", ", ".join(right) or "-", via)
        for union, left, right, via in rows
    ]
    widths = [max(len(row[i]) for row in (header, *cells)) for i in range(4)]
    lines = []
    for position, row in enumerate((header, *cells)):
        line = "  ".join(
            cell.ljust(width) for cell, width in zip(row, widths, strict=True)
        ).rstrip()
        if position and row[0] in marked:
            line = f"{line.ljust(sum(widths) + 6)}  <- conflict"
        lines.append(line)
    return lines

resolve_naming(op, *, left, right, canonical_maps=())

:func:build_naming, raising every blocking finding as one :class:MergeNamingError.

Source code in graflo/architecture/evolution/naming_graph.py

def resolve_naming(
    op: MergeManifestsOp,
    *,
    left: GraphManifest,
    right: GraphManifest,
    canonical_maps: Sequence[tuple[Side, CanonicalMap]] = (),
) -> ClusterResolution:
    """:func:`build_naming`, raising every blocking finding as one :class:`MergeNamingError`."""
    result = build_naming(op, left=left, right=right, canonical_maps=canonical_maps)
    result.raise_if_blocking()
    return result.resolution

suggest_merge_op(left, right, op=None, *, canonical_maps=(), max_steps=32)

A merge op that settles what can be settled without guessing, for review.

Without op, a scaffold: an equivalence for every name both sides share. With one, op with the first repair of each blocking finding applied — a rename away before a new into, and either before fusing entities — until no finding has one left, and each automatic own key written out where the identity is declared. Two spellings of one concept are never unioned: that is a guess, left to the author. Nothing is applied to a manifest; the result is a value to read, edit and save.

Source code in graflo/architecture/evolution/naming_graph.py

def suggest_merge_op(
    left: GraphManifest,
    right: GraphManifest,
    op: MergeManifestsOp | None = None,
    *,
    canonical_maps: Sequence[tuple[Side, CanonicalMap]] = (),
    max_steps: int = 32,
) -> MergeManifestsOp:
    """A merge op that settles what can be settled without guessing, for review.

    Without *op*, a scaffold: an equivalence for every name both sides share.
    With one, *op* with the first repair of each blocking finding applied —
    a rename away before a new ``into``, and either before fusing entities —
    until no finding has one left, and each automatic own key written out
    where the identity is declared. Two spellings of one concept are never
    unioned: that is a guess, left to the author. Nothing is applied to a
    manifest; the result is a value to read, edit and save.
    """
    current = op if op is not None else MergeManifestsOp()
    tried: set[str] = set()
    for _ in range(max_steps):
        result = build_naming(
            current, left=left, right=right, canonical_maps=canonical_maps
        )
        progressed = False
        for finding in result.blocking:
            if finding.kind == "near_collision":
                continue
            for repair in finding.repairs:
                key = repr(repair.to_dict())
                if key in tried:
                    continue
                tried.add(key)
                repaired = apply_repair(current, repair)
                if repaired is None:
                    continue
                current, progressed = repaired, True
                break
            if progressed:
                break
        if not progressed:
            break
    result = build_naming(
        current, left=left, right=right, canonical_maps=canonical_maps
    )
    return _with_explicit_keys(current, result)