graflo.architecture.evolution.equivalence

Equivalence clusters over merge-time vertex/relation mappings.

A :class:~graflo.architecture.evolution.ops.VertexEquivalence (or :class:~graflo.architecture.evolution.ops.RelationEquivalence) declares one n-ary cluster directly: left / right name one or more members on each side, collapsing onto one merged name. :func:index_clusters is the consistency check over the declared clusters of one :class:~graflo.architecture.evolution.ops.MergeManifestsOp — there is no connected-component search left to do (one declaration is one cluster); it validates that the declarations do not overlap or collapse into each other by accident:

• no (side, name) may be claimed by two declarations — that is the author's job to state as one cluster, not two;
• two declarations must not share one merged name — sharing one collapses them into one merged class, which must be spelled as one n-ary cluster so it is visible to review, not left implicit;
• a merged name that already exists as a different, non-member class on a side must not be silently merged into — add it to the cluster explicitly. The one exception is a name that another declaration renames away: the lowered map applies in one step, so the side lands on a vacated name whether it is one member or a merge.

Members and merged names are resolved before indexing — into may be omitted and a member may be spelled by its canonical name — by :func:~graflo.architecture.evolution.canonical.resolve_clusters, which hands the resolved shapes in as :class:ClusterSpec s. Nodes are (side, name) pairs so a class named Org on the left is never confused with Org on the right.

Attributes

DeclarationT = TypeVar('DeclarationT', VertexEquivalence, RelationEquivalence) module-attribute

Kind = Literal['vertex', 'relation'] module-attribute

RelationCluster = Cluster[RelationEquivalence] module-attribute

Side = Literal['left', 'right'] module-attribute

SubjectScope = Literal['left', 'right', 'merged', 'canonical'] module-attribute

Classes

Cluster dataclass

Bases: Generic[DeclarationT]

One n-ary equivalence cluster, over vertices or relations, in resolved names.

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

@dataclass(frozen=True)
class Cluster(Generic[DeclarationT]):
    """One n-ary equivalence cluster, over vertices or relations, in resolved names."""

    left: tuple[str, ...]
    right: tuple[str, ...]
    into: str
    declaration: DeclarationT
    aliases: dict[Side, dict[str, str]] = field(default_factory=dict)
    declared_into: str | None = None
    synthesized: bool = False

    def members(self, side: Side) -> tuple[str, ...]:
        return self.left if side == "left" else self.right

    def resolved(self, side: Side, declared: str) -> str:
        """The member a spelling names on *side*: its own name, or its canonical one."""
        return self.aliases.get(side, {}).get(declared, declared)

    def property_maps(self, side: Side) -> dict[str, dict[str, str]]:
        """The declaration's per-member attribute maps, keyed by resolved member name."""
        declaration = self.declaration
        if not isinstance(declaration, VertexEquivalence):
            return {}
        return {
            self.resolved(side, member): attrs
            for member, attrs in declaration.property_maps(side).items()
        }

Attributes

aliases = field(default_factory=dict) class-attribute instance-attribute

declaration instance-attribute

declared_into = None class-attribute instance-attribute

into instance-attribute

left instance-attribute

right instance-attribute

synthesized = False class-attribute instance-attribute

Methods:

__init__(left, right, into, declaration, aliases=dict(), declared_into=None, synthesized=False)

members(side)

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

def members(self, side: Side) -> tuple[str, ...]:
    return self.left if side == "left" else self.right

property_maps(side)

The declaration's per-member attribute maps, keyed by resolved member name.

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

def property_maps(self, side: Side) -> dict[str, dict[str, str]]:
    """The declaration's per-member attribute maps, keyed by resolved member name."""
    declaration = self.declaration
    if not isinstance(declaration, VertexEquivalence):
        return {}
    return {
        self.resolved(side, member): attrs
        for member, attrs in declaration.property_maps(side).items()
    }

resolved(side, declared)

The member a spelling names on side: its own name, or its canonical one.

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

def resolved(self, side: Side, declared: str) -> str:
    """The member a spelling names on *side*: its own name, or its canonical one."""
    return self.aliases.get(side, {}).get(declared, declared)

ClusterConflictError

Bases: Refusal

Two or more equivalence declarations conflict over cluster membership.

check names the rule that refused and subjects the names it is about, as :func:subject ids -- see :class:.Refusal.

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

class ClusterConflictError(Refusal):
    """Two or more equivalence declarations conflict over cluster membership.

    ``check`` names the rule that refused and ``subjects`` the names it is
    about, as :func:`subject` ids -- see :class:`.Refusal`.
    """

ClusterIndex dataclass

Every declared cluster of one merge op, validated for consistency.

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

@dataclass(frozen=True)
class ClusterIndex:
    """Every declared cluster of one merge op, validated for consistency."""

    vertices: tuple[Cluster[VertexEquivalence], ...]
    relations: tuple[Cluster[RelationEquivalence], ...]

    @property
    def labels(self) -> frozenset[str]:
        """The merged names of every vertex cluster."""
        return frozenset(c.into for c in self.vertices)

    @property
    def relation_labels(self) -> frozenset[str]:
        """The merged names of every relation cluster."""
        return frozenset(c.into for c in self.relations)

    @property
    def declared_intos(self) -> frozenset[str]:
        """Every merged name as the author spelled it, before translation."""
        return frozenset(
            c.declared_into
            for c in (*self.vertices, *self.relations)
            if c.declared_into is not None
        )

    def vertex_members(self, side: Side) -> frozenset[str]:
        out: set[str] = set()
        for c in self.vertices:
            out.update(c.members(side))
        return frozenset(out)

    def relation_members(self, side: Side) -> frozenset[str]:
        out: set[str] = set()
        for c in self.relations:
            out.update(c.members(side))
        return frozenset(out)

    def cluster_for_label(self, into: str) -> Cluster[VertexEquivalence] | None:
        """The vertex cluster collapsing onto *into*, or ``None``."""
        return next((c for c in self.vertices if c.into == into), None)

Attributes

declared_intos property

Every merged name as the author spelled it, before translation.

labels property

The merged names of every vertex cluster.

relation_labels property

The merged names of every relation cluster.

relations instance-attribute

vertices instance-attribute

Methods:

__init__(vertices, relations)

cluster_for_label(into)

The vertex cluster collapsing onto into, or None.

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

def cluster_for_label(self, into: str) -> Cluster[VertexEquivalence] | None:
    """The vertex cluster collapsing onto *into*, or ``None``."""
    return next((c for c in self.vertices if c.into == into), None)

relation_members(side)

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

def relation_members(self, side: Side) -> frozenset[str]:
    out: set[str] = set()
    for c in self.relations:
        out.update(c.members(side))
    return frozenset(out)

vertex_members(side)

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

def vertex_members(self, side: Side) -> frozenset[str]:
    out: set[str] = set()
    for c in self.vertices:
        out.update(c.members(side))
    return frozenset(out)

ClusterSpec dataclass

One declaration's resolved shape: members in the manifests' own names, and its merged name.

aliases records, per side, every other name a member answers to — the canonical name it was declared by, or the one the canonical map gives it — so the per-member maps (property equivalences, SideIdentity.members, identity-alignment member keys) may be keyed by either the member's own name or its canonical one. declared_into is the merged name as the author spelled it, before any canonical map translated it; synthesized marks a cluster merge created itself for a same-name pair under name_conflict="union_right".

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

@dataclass(frozen=True)
class ClusterSpec:
    """One declaration's resolved shape: members in the manifests' own names, and its merged name.

    ``aliases`` records, per side, every other name a member answers to —
    the canonical name it was declared by, or the one the canonical map gives
    it — so the per-member maps (property equivalences,
    ``SideIdentity.members``, identity-alignment member keys) may be keyed by
    either the member's own name or its canonical one. ``declared_into`` is
    the merged name as the author spelled it, before any canonical map
    translated it; ``synthesized`` marks a cluster merge created itself for
    a same-name pair under ``name_conflict="union_right"``.
    """

    left: tuple[str, ...]
    right: tuple[str, ...]
    into: str
    aliases: dict[Side, dict[str, str]] = field(default_factory=dict)
    declared_into: str | None = None
    synthesized: bool = False

Attributes

aliases = field(default_factory=dict) class-attribute instance-attribute

declared_into = None class-attribute instance-attribute

into instance-attribute

left instance-attribute

right instance-attribute

synthesized = False class-attribute instance-attribute

Methods:

__init__(left, right, into, aliases=dict(), declared_into=None, synthesized=False)

UnknownMemberError

Bases: Refusal

An equivalence names a member the manifest on that side does not declare.

Its own type because it is the one refusal here that is nearly always a typo rather than a disagreement between two declarations, and a caller classifying refusals cannot key on a bare ValueError.

Derives check and subjects rather than taking them from the caller: there is only one rule it can be an instance of, and only one name it can be about.

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

class UnknownMemberError(Refusal):
    """An equivalence names a member the manifest on that side does not declare.

    Its own type because it is the one refusal here that is nearly always a
    typo rather than a disagreement between two declarations, and a caller
    classifying refusals cannot key on a bare ``ValueError``.

    Derives ``check`` and ``subjects`` rather than taking them from the caller:
    there is only one rule it can be an instance of, and only one name it can
    be about.
    """

    def __init__(self, message: str, *, side: Side, kind: Kind, member: str) -> None:
        super().__init__(
            message,
            check=f"unknown {kind} member",
            subjects=(subject(side, member),),
        )
        self.side = side
        self.kind = kind
        self.member = member

Attributes

kind = kind instance-attribute

member = member instance-attribute

side = side instance-attribute

Methods:

__init__(message, *, side, kind, member)

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

def __init__(self, message: str, *, side: Side, kind: Kind, member: str) -> None:
    super().__init__(
        message,
        check=f"unknown {kind} member",
        subjects=(subject(side, member),),
    )
    self.side = side
    self.kind = kind
    self.member = member

Functions:

check_member_existence(vertex_clusters, relation_clusters, *, left_vertex_names, right_vertex_names, left_relation_names, right_relation_names)

Every member must exist on its side; a near-miss spelling is named.

Raises:

Type	Description
UnknownMemberError	A member is absent from its side. A subclass of ValueError, so existing handlers are unaffected.

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

def check_member_existence(
    vertex_clusters: Iterable[ClusterSpec | Cluster[VertexEquivalence]],
    relation_clusters: Iterable[ClusterSpec | Cluster[RelationEquivalence]],
    *,
    left_vertex_names: Collection[str],
    right_vertex_names: Collection[str],
    left_relation_names: Collection[str],
    right_relation_names: Collection[str],
) -> None:
    """Every member must exist on its side; a near-miss spelling is named.

    Raises:
        UnknownMemberError: A member is absent from its side. A subclass of
            ``ValueError``, so existing handlers are unaffected.
    """
    for cluster in vertex_clusters:
        for member in cluster.left:
            if member not in left_vertex_names:
                raise UnknownMemberError(
                    f"merge_manifests: left vertex {member!r} not in left "
                    f"manifest{did_you_mean(member, left_vertex_names)}",
                    side="left",
                    kind="vertex",
                    member=member,
                )
        for member in cluster.right:
            if member not in right_vertex_names:
                raise UnknownMemberError(
                    f"merge_manifests: right vertex {member!r} not in right "
                    f"manifest{did_you_mean(member, right_vertex_names)}",
                    side="right",
                    kind="vertex",
                    member=member,
                )
    for cluster in relation_clusters:
        for member in cluster.left:
            if member not in left_relation_names:
                raise UnknownMemberError(
                    f"merge_manifests: left relation {member!r} not in left manifest",
                    side="left",
                    kind="relation",
                    member=member,
                )
        for member in cluster.right:
            if member not in right_relation_names:
                raise UnknownMemberError(
                    f"merge_manifests: right relation {member!r} not in right manifest",
                    side="right",
                    kind="relation",
                    member=member,
                )

declared_spec(declaration)

The shape a declaration states outright, with no canonical map to consult.

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

def declared_spec(declaration: VertexEquivalence | RelationEquivalence) -> ClusterSpec:
    """The shape a declaration states outright, with no canonical map to consult."""
    if declaration.into is None:
        raise ValueError(
            f"equivalence {declaration.left_members} ~ "
            f"{declaration.right_members} has no `into`; a merged name comes "
            "from `into`, from a canonical map on the merge op, or from one "
            "spelling every member shares — resolve it through "
            "validate_and_complete_canonical_map, or name it"
        )
    return ClusterSpec(
        left=tuple(declaration.left_members),
        right=tuple(declaration.right_members),
        into=declaration.into,
        declared_into=declaration.into,
    )

did_you_mean(name, candidates)

A suffix naming a candidate that denotes the same concept, if any.

Authoring an equivalence in the wrong convention is the likeliest mistake at this boundary, and "not in left manifest" alone is a dead end when the vertex is right there under another spelling.

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

def did_you_mean(name: str, candidates: Iterable[str]) -> str:
    """A suffix naming a candidate that denotes the same concept, if any.

    Authoring an equivalence in the wrong convention is the likeliest mistake
    at this boundary, and "not in left manifest" alone is a dead end when the
    vertex is right there under another spelling.
    """
    key = canonical_slug(name)
    near = sorted(c for c in candidates if c != name and canonical_slug(c) == key)
    if not near:
        return ""
    return (
        f"; it has {near[0]!r}, which denotes the same concept — author the "
        "equivalence in the manifest's own spelling"
    )

index_clusters(op, *, left_vertices=(), right_vertices=(), left_relations=(), right_relations=(), vertex_specs=None, relation_specs=None)

Validate and index the declared clusters of op.

vertex_specs / relation_specs are the resolved shapes, aligned with the op's declaration lists; omitted, each declaration is taken as written (which requires into). The name collections are what a merged name may collide with on each side.

Raises :class:ClusterConflictError on an overlapping declaration, two declarations sharing one merged name, or a merged name that would silently occupy an existing non-member class on a side.

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

def index_clusters(
    op: MergeManifestsOp,
    *,
    left_vertices: Collection[str] = (),
    right_vertices: Collection[str] = (),
    left_relations: Collection[str] = (),
    right_relations: Collection[str] = (),
    vertex_specs: Sequence[ClusterSpec] | None = None,
    relation_specs: Sequence[ClusterSpec] | None = None,
) -> ClusterIndex:
    """Validate and index the declared clusters of *op*.

    *vertex_specs* / *relation_specs* are the resolved shapes, aligned with the
    op's declaration lists; omitted, each declaration is taken as written
    (which requires ``into``). The name collections are what a merged name
    may collide with on each side.

    Raises :class:`ClusterConflictError` on an overlapping declaration, two
    declarations sharing one merged name, or a merged name that would
    silently occupy an existing non-member class on a side.
    """
    if vertex_specs is None:
        vertex_specs = [declared_spec(v) for v in op.vertex_equivalences]
    if relation_specs is None:
        relation_specs = [declared_spec(r) for r in op.relation_equivalences]
    _check_declarations(
        vertex_specs,
        kind="vertex equivalence",
        left_names=left_vertices,
        right_names=right_vertices,
    )
    _check_declarations(
        relation_specs,
        kind="relation equivalence",
        left_names=left_relations,
        right_names=right_relations,
    )
    return ClusterIndex(
        vertices=tuple(
            Cluster(
                left=spec.left,
                right=spec.right,
                into=spec.into,
                declaration=v,
                aliases=spec.aliases,
                declared_into=spec.declared_into,
                synthesized=spec.synthesized,
            )
            for v, spec in zip(op.vertex_equivalences, vertex_specs, strict=True)
        ),
        relations=tuple(
            Cluster(
                left=spec.left,
                right=spec.right,
                into=spec.into,
                declaration=r,
                aliases=spec.aliases,
                declared_into=spec.declared_into,
                synthesized=spec.synthesized,
            )
            for r, spec in zip(op.relation_equivalences, relation_specs, strict=True)
        ),
    )

subject(scope, name, attr=None)

A stable id for the class or attribute a refusal is about.

Every refusal at this boundary names the declarations it refuses, and a caller that wants to point at them — a preview, a diagram, an editor — needs those names as data rather than parsed back out of prose.

Parameters:

Name	Type	Description	Default
scope	SubjectScope	Which vocabulary the name lives in.	required
name	str	The class or relation name.	required
attr	str | None	An attribute of it, when the subject is narrower than a class.	None

Returns:

Type	Description
str	"left:Firm" for a class, "left:Firm.firm_id" for an attribute.

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

def subject(scope: SubjectScope, name: str, attr: str | None = None) -> str:
    """A stable id for the class or attribute a refusal is about.

    Every refusal at this boundary names the declarations it refuses, and a
    caller that wants to *point* at them — a preview, a diagram, an editor —
    needs those names as data rather than parsed back out of prose.

    Args:
        scope: Which vocabulary the name lives in.
        name: The class or relation name.
        attr: An attribute of it, when the subject is narrower than a class.

    Returns:
        ``"left:Firm"`` for a class, ``"left:Firm.firm_id"`` for an attribute.
    """
    return f"{scope}:{name}" if attr is None else f"{scope}:{name}.{attr}"