Skip to content

graflo.architecture.evolution.compare

Two versions of a schema, side by side.

:func:compare_schemas lines up the vertex types, relations and fields of two manifests as rows -- one row per element, renamed or not -- with each side's values. Given their common ancestor, rows are anchored on it and each side's recorded renames (:class:RenameHints, ancestor → side) say where an ancestor element went; without one, rows pair by name, then by canonical name.

Every row carries the slots :func:merge_three_way addresses it by, so :meth:SchemaComparison.locate finds the row a merge conflict is about.

A derived view of the schema, not part of the contract: nothing here is hashed or stored.

Attributes

CompareStatus = Literal['unchanged', 'added', 'removed', 'renamed', 'changed'] module-attribute

A row read left → right: added is only on the right, removed only on the left.

EdgeKey = tuple[str, str, str | None] module-attribute

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

__all__ = ['CompareFocus', 'CompareStatus', 'FieldRow', 'FieldSide', 'RelationRow', 'RelationSide', 'SchemaComparison', 'VertexRow', 'VertexSide', 'compare_schemas'] module-attribute

Classes

CompareFocus

Bases: ConfigBaseModel

Where a merge slot lands among the rows.

Source code in graflo/architecture/evolution/compare.py 
139
140
141
142
143
144
145
146
147
class CompareFocus(ConfigBaseModel):
    """Where a merge slot lands among the rows."""

    row_id: str
    field_row_id: str | None = None
    aspect: str | None = PydanticField(
        default=None,
        description="The part of the row the slot names: description, identity, fields, ...",
    )

Attributes

aspect = PydanticField(default=None, description='The part of the row the slot names: description, identity, fields, ...') class-attribute instance-attribute
field_row_id = None class-attribute instance-attribute
row_id instance-attribute

FieldRow

Bases: ConfigBaseModel

A field (vertex property or edge property) on both sides.

Source code in graflo/architecture/evolution/compare.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class FieldRow(ConfigBaseModel):
    """A field (vertex property or edge property) on both sides."""

    id: str
    left: FieldSide | None = None
    right: FieldSide | None = None
    status: CompareStatus
    aspects: list[str] = PydanticField(
        default_factory=list,
        description="What differs between the sides: name, type, description, identity.",
    )
    changed_on: list[Side] = PydanticField(
        default_factory=list,
        description="The sides that changed it since the common ancestor.",
    )
    slots: list[list[str]] = PydanticField(default_factory=list)

Attributes

aspects = PydanticField(default_factory=list, description='What differs between the sides: name, type, description, identity.') class-attribute instance-attribute
changed_on = PydanticField(default_factory=list, description='The sides that changed it since the common ancestor.') class-attribute instance-attribute
id instance-attribute
left = None class-attribute instance-attribute
right = None class-attribute instance-attribute
slots = PydanticField(default_factory=list) class-attribute instance-attribute
status instance-attribute

FieldSide

Bases: ConfigBaseModel

One side of a field row.

Source code in graflo/architecture/evolution/compare.py 
47
48
49
50
51
52
53
54
55
56
57
class FieldSide(ConfigBaseModel):
    """One side of a field row."""

    name: str
    type: str | None = PydanticField(
        default=None, description='``"INT"``, ``"LIST[STRING]"``; None when untyped.'
    )
    description: str | None = None
    in_identity: bool = PydanticField(
        default=False, description="Whether the field is part of the primary key."
    )

Attributes

description = None class-attribute instance-attribute
in_identity = PydanticField(default=False, description='Whether the field is part of the primary key.') class-attribute instance-attribute
name instance-attribute
type = PydanticField(default=None, description='``"INT"``, ``"LIST[STRING]"``; None when untyped.') class-attribute instance-attribute

RelationRow

Bases: ConfigBaseModel

An edge on both sides, with its properties.

Rows are keyed by their endpoint rows, so an edge whose endpoint was renamed stays on one row; an edge moved to other endpoints is a removed and an added row.

Source code in graflo/architecture/evolution/compare.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class RelationRow(ConfigBaseModel):
    """An edge on both sides, with its properties.

    Rows are keyed by their endpoint rows, so an edge whose endpoint was renamed
    stays on one row; an edge moved to other endpoints is a removed and an added row.
    """

    id: str
    left: RelationSide | None = None
    right: RelationSide | None = None
    status: CompareStatus
    aspects: list[str] = PydanticField(
        default_factory=list,
        description="What differs: name, description, directed, identity, fields.",
    )
    fields: list[FieldRow] = PydanticField(default_factory=list)
    changed_on: list[Side] = PydanticField(default_factory=list)
    slots: list[list[str]] = PydanticField(default_factory=list)

Attributes

aspects = PydanticField(default_factory=list, description='What differs: name, description, directed, identity, fields.') class-attribute instance-attribute
changed_on = PydanticField(default_factory=list) class-attribute instance-attribute
fields = PydanticField(default_factory=list) class-attribute instance-attribute
id instance-attribute
left = None class-attribute instance-attribute
right = None class-attribute instance-attribute
slots = PydanticField(default_factory=list) class-attribute instance-attribute
status instance-attribute

RelationSide

Bases: ConfigBaseModel

One side of a relation row: one edge.

Source code in graflo/architecture/evolution/compare.py 
106
107
108
109
110
111
112
113
114
115
116
class RelationSide(ConfigBaseModel):
    """One side of a relation row: one edge."""

    relation: str | None = None
    source: str
    target: str
    source_row: str = PydanticField(description="The id of the source's vertex row.")
    target_row: str = PydanticField(description="The id of the target's vertex row.")
    description: str | None = None
    directed: bool = True
    identities: list[list[str]] = PydanticField(default_factory=list)

Attributes

description = None class-attribute instance-attribute
directed = True class-attribute instance-attribute
identities = PydanticField(default_factory=list) class-attribute instance-attribute
relation = None class-attribute instance-attribute
source instance-attribute
source_row = PydanticField(description="The id of the source's vertex row.") class-attribute instance-attribute
target instance-attribute
target_row = PydanticField(description="The id of the target's vertex row.") class-attribute instance-attribute

SchemaComparison

Bases: ConfigBaseModel

Two schemas as aligned rows.

Source code in graflo/architecture/evolution/compare.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
class SchemaComparison(ConfigBaseModel):
    """Two schemas as aligned rows."""

    left_label: str | None = None
    right_label: str | None = None
    base_label: str | None = None
    aligned_on_base: bool = PydanticField(
        description="Rows are anchored on the common ancestor (else paired by name)."
    )
    vertices: list[VertexRow] = PydanticField(default_factory=list)
    relations: list[RelationRow] = PydanticField(default_factory=list)
    counts: dict[str, int] = PydanticField(
        default_factory=dict, description="Vertex and relation rows per status."
    )
    warnings: list[str] = PydanticField(default_factory=list)

    def locate(self, slot: Sequence[str]) -> CompareFocus | None:
        """The row (and field row) a merge slot addresses, or None.

        The deepest row slot that covers *slot* wins; what *slot* says beyond it
        is the aspect (``("vertex", "company", "description")`` → the Company row,
        aspect ``description``). Match on the slot, never on its label: a
        canonical name may itself contain ``/``.
        """
        target: Slot = tuple(slot)
        best: tuple[int, CompareFocus] | None = None

        def consider(row_id: str, field_id: str | None, slots: list[list[str]]):
            nonlocal best
            for candidate in slots:
                outer = tuple(candidate)
                if not _covers(outer, target):
                    continue
                if best is not None and len(outer) <= best[0]:
                    continue
                rest = target[len(outer) :]
                aspect = rest[0] if rest else None
                if aspect == "field":
                    aspect = "fields"
                best = (
                    len(outer),
                    CompareFocus(row_id=row_id, field_row_id=field_id, aspect=aspect),
                )

        for row in [*self.vertices, *self.relations]:
            consider(row.id, None, row.slots)
            for field_row in row.fields:
                consider(row.id, field_row.id, field_row.slots)
        return None if best is None else best[1]

Attributes

aligned_on_base = PydanticField(description='Rows are anchored on the common ancestor (else paired by name).') class-attribute instance-attribute
base_label = None class-attribute instance-attribute
counts = PydanticField(default_factory=dict, description='Vertex and relation rows per status.') class-attribute instance-attribute
left_label = None class-attribute instance-attribute
relations = PydanticField(default_factory=list) class-attribute instance-attribute
right_label = None class-attribute instance-attribute
vertices = PydanticField(default_factory=list) class-attribute instance-attribute
warnings = PydanticField(default_factory=list) class-attribute instance-attribute

Methods:

locate(slot)

The row (and field row) a merge slot addresses, or None.

The deepest row slot that covers slot wins; what slot says beyond it is the aspect (("vertex", "company", "description") → the Company row, aspect description). Match on the slot, never on its label: a canonical name may itself contain /.

Source code in graflo/architecture/evolution/compare.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def locate(self, slot: Sequence[str]) -> CompareFocus | None:
    """The row (and field row) a merge slot addresses, or None.

    The deepest row slot that covers *slot* wins; what *slot* says beyond it
    is the aspect (``("vertex", "company", "description")`` → the Company row,
    aspect ``description``). Match on the slot, never on its label: a
    canonical name may itself contain ``/``.
    """
    target: Slot = tuple(slot)
    best: tuple[int, CompareFocus] | None = None

    def consider(row_id: str, field_id: str | None, slots: list[list[str]]):
        nonlocal best
        for candidate in slots:
            outer = tuple(candidate)
            if not _covers(outer, target):
                continue
            if best is not None and len(outer) <= best[0]:
                continue
            rest = target[len(outer) :]
            aspect = rest[0] if rest else None
            if aspect == "field":
                aspect = "fields"
            best = (
                len(outer),
                CompareFocus(row_id=row_id, field_row_id=field_id, aspect=aspect),
            )

    for row in [*self.vertices, *self.relations]:
        consider(row.id, None, row.slots)
        for field_row in row.fields:
            consider(row.id, field_row.id, field_row.slots)
    return None if best is None else best[1]

VertexRow

Bases: ConfigBaseModel

A vertex type on both sides, with its fields.

Source code in graflo/architecture/evolution/compare.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class VertexRow(ConfigBaseModel):
    """A vertex type on both sides, with its fields."""

    id: str
    left: VertexSide | None = None
    right: VertexSide | None = None
    status: CompareStatus
    aspects: list[str] = PydanticField(
        default_factory=list,
        description="What differs: name, description, identity, secondary, fields.",
    )
    fields: list[FieldRow] = PydanticField(default_factory=list)
    changed_on: list[Side] = PydanticField(default_factory=list)
    slots: list[list[str]] = PydanticField(default_factory=list)

Attributes

aspects = PydanticField(default_factory=list, description='What differs: name, description, identity, secondary, fields.') class-attribute instance-attribute
changed_on = PydanticField(default_factory=list) class-attribute instance-attribute
fields = PydanticField(default_factory=list) class-attribute instance-attribute
id instance-attribute
left = None class-attribute instance-attribute
right = None class-attribute instance-attribute
slots = PydanticField(default_factory=list) class-attribute instance-attribute
status instance-attribute

VertexSide

Bases: ConfigBaseModel

One side of a vertex row.

Source code in graflo/architecture/evolution/compare.py 
78
79
80
81
82
83
84
85
86
87
class VertexSide(ConfigBaseModel):
    """One side of a vertex row."""

    name: str
    description: str | None = None
    identity: list[str] = PydanticField(default_factory=list)
    identity_mode: str = PydanticField(
        description="natural, hash, funnel, assigned or blank."
    )
    secondary_identities: list[list[str]] = PydanticField(default_factory=list)

Attributes

description = None class-attribute instance-attribute
identity = PydanticField(default_factory=list) class-attribute instance-attribute
identity_mode = PydanticField(description='natural, hash, funnel, assigned or blank.') class-attribute instance-attribute
name instance-attribute
secondary_identities = PydanticField(default_factory=list) class-attribute instance-attribute

Functions:

compare_schemas(left, right, *, base=None, left_hints=None, right_hints=None, hints=None, left_label=None, right_label=None, base_label=None)

Align the schemas of left and right into rows.

Parameters:

Name Type Description Default
left GraphManifest

One version.

 required
right GraphManifest

The other version.

 required
base GraphManifest | None

Their common ancestor. Rows are anchored on it, and each row's changed_on says which side changed it.

 None
left_hints RenameHints | None

Renames from base to left, as :func:merge_three_way takes them (:meth:RenameHints.from_ops over that side's ops).

 None
right_hints RenameHints | None

Renames from base to right.

 None
hints RenameHints | None

Renames from left to right; read only without base.

 None
left_label str | None

The left side's name (a version), echoed back.

 None
right_label str | None

The right side's name.

 None
base_label str | None

The ancestor's name.

 None

Returns:

Type Description
SchemaComparison

The rows in declaration order: the ancestor's elements first, then what
SchemaComparison

the left side added, then what only the right side has.
Source code in graflo/architecture/evolution/compare.py 
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
def compare_schemas(
    left: GraphManifest,
    right: GraphManifest,
    *,
    base: GraphManifest | None = None,
    left_hints: RenameHints | None = None,
    right_hints: RenameHints | None = None,
    hints: RenameHints | None = None,
    left_label: str | None = None,
    right_label: str | None = None,
    base_label: str | None = None,
) -> SchemaComparison:
    """Align the schemas of *left* and *right* into rows.

    Args:
        left: One version.
        right: The other version.
        base: Their common ancestor. Rows are anchored on it, and each row's
            ``changed_on`` says which side changed it.
        left_hints: Renames from *base* to *left*, as :func:`merge_three_way`
            takes them (:meth:`RenameHints.from_ops` over that side's ops).
        right_hints: Renames from *base* to *right*.
        hints: Renames from *left* to *right*; read only without *base*.
        left_label: The left side's name (a version), echoed back.
        right_label: The right side's name.
        base_label: The ancestor's name.

    Returns:
        The rows in declaration order: the ancestor's elements first, then what
        the left side added, then what only the right side has.
    """
    ctx = _Context(
        left,
        right,
        base,
        left_hints or RenameHints(),
        right_hints or RenameHints(),
        hints or RenameHints(),
    )
    warnings: list[str] = []
    if base is None and hints is None:
        warnings.append(
            "No common ancestor: rows are paired by name, so a rename shows as "
            "one removed and one added row."
        )
    vertex_rows = _vertex_rows(ctx)
    relation_rows = _relation_rows(ctx)
    counts: dict[str, int] = {}
    for row in [*vertex_rows, *relation_rows]:
        counts[row.status] = counts.get(row.status, 0) + 1
    return SchemaComparison(
        left_label=left_label,
        right_label=right_label,
        base_label=base_label if base is not None else None,
        aligned_on_base=base is not None,
        vertices=vertex_rows,
        relations=relation_rows,
        counts=counts,
        warnings=warnings,
    )