graflo.architecture.evolution.ops

Typed manifest evolution operations.

Two field names recur across the models and mean two different things: into is where existing names collapse — the target of a merge, a rename, or an equivalence (MergeVerticesOp, MergeEdgesOp, VertexEquivalence, RelationEquivalence, PropertyEquivalence); name is what a new thing is called — an attribute an alignment derives (AlignmentAttribute, LocalKeySpec). A model never uses one for the other.

Attributes

INGESTION_REWRITING_OPS = frozenset({'add_inverse_edges', 'add_resource_transforms', 'add_resources', 'canonicalize', 'remove_resources', 'ensure_extracted_fields', 'merge_edges', 'merge_vertices', 'project_manifest', 'remove_edge_properties', 'remove_edges', 'remove_vertex_properties', 'remove_vertices', 'rename_edge_properties', 'rename_relations', 'rename_resources', 'rename_vertex_properties', 'rename_vertices', 'replace_identity', 'retarget_edges', 'sanitize', 'set_inverse_emission'}) module-attribute

IdentityBranchSpec = str | list[str] module-attribute

IdentityTarget = Annotated[NaturalIdentityTarget | HashIdentityTarget | FunnelIdentityTarget | AssignedIdentityTarget | BlankIdentityTarget, PydanticField(discriminator='mode')] module-attribute

ManifestOp = Annotated[RemoveVerticesOp | AddResourceTransformsOp | EnsureExtractedFieldsOp | AddResourcesOp | RemoveResourcesOp | AddVerticesOp | AddEdgesOp | RetargetEdgesOp | AddSecondaryIdentitiesOp | RemoveSecondaryIdentitiesOp | ReplaceEdgeIdentitiesOp | ChangeFieldTypesOp | AddVertexIndexesOp | RemoveVertexIndexesOp | AddEdgeIndexesOp | RemoveEdgeIndexesOp | SetEdgeDirectedOp | SetBindingsOp | SetDbProfileOp | SetVertexSemanticsOp | SetVertexDescriptionsOp | SetEdgeSemanticsOp | SetFieldSemanticsOp | MergeVerticesOp | CanonicalizeOp | RenameVertexPropertiesOp | RemoveVertexPropertiesOp | AddVertexPropertiesOp | RenameVerticesOp | RenameRelationsOp | RenameResourcesOp | RemoveEdgesOp | MergeEdgesOp | RenameEdgePropertiesOp | RemoveEdgePropertiesOp | AddEdgePropertiesOp | DeclareEdgeInversesOp | RetractEdgeInversesOp | AddInverseEdgesOp | SetNativeInversesOp | SetInverseEmissionOp | ProjectManifestOp | ReplaceIdentityOp | SanitizeOp | MergeManifestsOp, PydanticField(discriminator='op')] module-attribute

Classes

AddEdgeIndexesOp

Bases: ConfigBaseModel

Author secondary indexes on edge physical specs.

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

class AddEdgeIndexesOp(ConfigBaseModel):
    """Author secondary indexes on edge physical specs."""

    op: Literal["add_edge_indexes"] = "add_edge_indexes"
    edges: list[EdgeIndexEntry] = PydanticField(
        ...,
        description="Per-spec indexes to add (``indexes`` on each entry).",
        min_length=1,
    )

    @model_validator(mode="after")
    def _validate_entries(self) -> AddEdgeIndexesOp:
        _validate_edge_index_entries(
            self.edges, kind="add_edge_indexes", carries="indexes"
        )
        return self

Attributes

edges = PydanticField(..., description='Per-spec indexes to add (``indexes`` on each entry).', min_length=1) class-attribute instance-attribute

op = 'add_edge_indexes' class-attribute instance-attribute

AddEdgePropertiesOp

Bases: ConfigBaseModel

Add edge properties for each relation in schema/profile defaults.

An entry may be a bare name (an untyped property) or a full :class:~graflo.architecture.schema.vertex.Field, as on :class:AddVertexPropertiesOp, so a typed or grounded edge property is one replayable step rather than an add followed by a type change.

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

class AddEdgePropertiesOp(ConfigBaseModel):
    """Add edge properties for each relation in schema/profile defaults.

    An entry may be a bare name (an untyped property) or a full
    :class:`~graflo.architecture.schema.vertex.Field`, as on
    :class:`AddVertexPropertiesOp`, so a typed or grounded edge property is
    one replayable step rather than an add followed by a type change.
    """

    op: Literal["add_edge_properties"] = "add_edge_properties"
    additions: dict[str, list[str | Field]] = PydanticField(
        ...,
        description=(
            "Per-relation edge property additions: "
            "``{relation_name: [field_name | Field, ...]}``."
        ),
        min_length=1,
    )

    def field_names(self, relation: str) -> list[str]:
        """The names added to *relation*, whichever shape they were written in."""
        return [
            entry if isinstance(entry, str) else entry.name
            for entry in self.additions.get(relation, [])
        ]

Attributes

additions = PydanticField(..., description='Per-relation edge property additions: ``{relation_name: [field_name | Field, ...]}``.', min_length=1) class-attribute instance-attribute

op = 'add_edge_properties' class-attribute instance-attribute

Methods:

field_names(relation)

The names added to relation, whichever shape they were written in.

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

def field_names(self, relation: str) -> list[str]:
    """The names added to *relation*, whichever shape they were written in."""
    return [
        entry if isinstance(entry, str) else entry.name
        for entry in self.additions.get(relation, [])
    ]

AddEdgesOp

Bases: ConfigBaseModel

Introduce new logical edge relations between existing vertex types.

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

class AddEdgesOp(ConfigBaseModel):
    """Introduce new logical edge relations between existing vertex types."""

    op: Literal["add_edges"] = "add_edges"
    edges: list[Edge] = PydanticField(
        ...,
        description="Full edge definitions, in the shape the schema block accepts.",
        min_length=1,
    )

    @model_validator(mode="after")
    def _validate_unique_ids(self) -> AddEdgesOp:
        edge_ids = [(edge.source, edge.target, edge.relation) for edge in self.edges]
        if len(edge_ids) != len(set(edge_ids)):
            raise ValueError(
                "add_edges entries must be unique by (source, target, relation)"
            )
        return self

Attributes

edges = PydanticField(..., description='Full edge definitions, in the shape the schema block accepts.', min_length=1) class-attribute instance-attribute

op = 'add_edges' class-attribute instance-attribute

AddInverseEdgesOp

Bases: ConfigBaseModel

Realize declared inverse pairs as explicit logical edges (portable to every backend).

Edges are derived from the relation map: for each directed edge (S, T, r) whose relation has a declared pair inv, adds the logical edge (T, S, inv) unless it exists and its physical spec, and sets emit_inverse on the edge steps that write r, so the same rows write both. No step is generated: the inverse is mirrored at assembly, after the relation is resolved, which covers every way a step can name its relation. A resource that already writes the inverse with a step of its own is left alone. The pair must be declared first (:class:DeclareEdgeInversesOp); this op never declares. A symmetric relation has no inverse edge to add -- its edges are undirected. Refused for a relation whose inverse is native (:class:SetNativeInversesOp), since both would store the same fact.

Withdraw with :class:RemoveEdgesOp on the inverse edges: removing an edge clears the emit_inverse flags that fed it.

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

class AddInverseEdgesOp(ConfigBaseModel):
    """Realize declared inverse pairs as explicit logical edges (portable to every backend).

    Edges are derived from the relation map: for each directed edge ``(S, T, r)``
    whose relation has a declared pair ``inv``, adds the logical edge
    ``(T, S, inv)`` unless it exists and its physical spec, and sets
    ``emit_inverse`` on the edge steps that write ``r``, so the same rows write
    both. No step is generated: the inverse is mirrored at assembly, after the
    relation is resolved, which covers every way a step can name its relation.
    A resource that already writes the inverse with a step of its own is left
    alone. The pair must be declared first (:class:`DeclareEdgeInversesOp`);
    this op never declares. A symmetric relation has no inverse edge to add --
    its edges are undirected. Refused for a relation whose inverse is native
    (:class:`SetNativeInversesOp`), since both would store the same fact.

    Withdraw with :class:`RemoveEdgesOp` on the inverse edges: removing an edge
    clears the ``emit_inverse`` flags that fed it.
    """

    op: Literal["add_inverse_edges"] = "add_inverse_edges"
    relations: list[str] | None = PydanticField(
        default=None,
        description=(
            "Paired relations whose edges get their inverse edge. Omitted: every "
            "relation in ``edge_config.inverses``, both sides."
        ),
        min_length=1,
    )

    @model_validator(mode="after")
    def _validate_unique(self) -> AddInverseEdgesOp:
        if self.relations is not None and len(set(self.relations)) != len(
            self.relations
        ):
            raise ValueError("add_inverse_edges: relations must be unique")
        return self

Attributes

op = 'add_inverse_edges' class-attribute instance-attribute

relations = PydanticField(default=None, description='Paired relations whose edges get their inverse edge. Omitted: every relation in ``edge_config.inverses``, both sides.', min_length=1) class-attribute instance-attribute

AddResourceTransformsOp

Bases: ConfigBaseModel

Append transform steps to named resources' pipelines.

The first op whose primary effect is ingestion: graph_schema is untouched. Steps land at the root level of each pipeline unless at names a deeper one; at whichever level they land, actor type-priority sorting (transform runs before vertex extraction at the same level) makes the position safe.

The level is load-bearing rather than cosmetic. An actor reads its transform buffer at its own LocationIndex with no ancestor fallback, and a descend subtree runs before its own level's transforms, so a step appended at the root is invisible to a vertex produced under a descend — and a transform whose declared inputs are missing skips silently by default. Target the level that produces the vertex.

Steps may reference a registry transform via call.use (resolved against the manifest's existing ingestion_model.transforms union the op's own transforms) or carry a fully inline call (module + foo + params), which cannot collide by name.

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

class AddResourceTransformsOp(ConfigBaseModel):
    """Append transform steps to named resources' pipelines.

    The first op whose primary effect is ingestion: ``graph_schema`` is
    untouched. Steps land at the root level of each pipeline unless ``at``
    names a deeper one; at whichever level they land, actor type-priority
    sorting (transform runs before vertex extraction at the same level) makes
    the position safe.

    The level is load-bearing rather than cosmetic. An actor reads its
    transform buffer at its own ``LocationIndex`` with no ancestor fallback,
    and a ``descend`` subtree runs *before* its own level's transforms, so a
    step appended at the root is invisible to a vertex produced under a
    ``descend`` — and a transform whose declared inputs are missing skips
    silently by default. Target the level that produces the vertex.

    Steps may reference a registry transform via ``call.use`` (resolved
    against the manifest's existing ``ingestion_model.transforms`` union the
    op's own ``transforms``) or carry a fully inline ``call``
    (``module`` + ``foo`` + ``params``), which cannot collide by name.
    """

    op: Literal["add_resource_transforms"] = "add_resource_transforms"
    additions: dict[str, list[dict[str, Any]]] = PydanticField(
        ...,
        description=(
            "Per-resource transform steps to append: "
            "``{resource_name: [step_dict, ...]}``."
        ),
        min_length=1,
    )
    at: dict[str, list[int]] = PydanticField(
        default_factory=dict,
        description=(
            "Per-resource pipeline level to append into: "
            "``{resource_name: [step_index, ...]}``. Each index must address a "
            "``descend`` step, descending one level per element; an omitted or "
            "empty path means the root level."
        ),
    )
    transforms: list[ProtoTransform] = PydanticField(
        default_factory=list,
        description=(
            "Named transforms to register in ``ingestion_model.transforms`` "
            "for steps that reference them via ``call.use``. A name already "
            "registered with a different body is an error at apply time."
        ),
    )

    @model_validator(mode="after")
    def _validate_steps(self) -> AddResourceTransformsOp:
        from graflo.architecture.contract.ingestion.steps.models import (
            TransformActorConfig,
        )
        from graflo.architecture.contract.ingestion.steps.normalize import (
            normalize_actor_step,
        )

        for resource_name, steps in self.additions.items():
            if not steps:
                raise ValueError(
                    f"add_resource_transforms: empty step list for resource "
                    f"{resource_name!r}"
                )
            for step in steps:
                normalized = normalize_actor_step(step)
                if (
                    not isinstance(normalized, dict)
                    or normalized.get("type") != "transform"
                ):
                    raise ValueError(
                        f"add_resource_transforms: step for resource "
                        f"{resource_name!r} is not a transform step: {step!r}"
                    )
                TransformActorConfig.model_validate(normalized)
        stray = sorted(set(self.at) - set(self.additions))
        if stray:
            raise ValueError(
                f"add_resource_transforms: `at` names resources {stray} with no "
                "steps to append"
            )
        for resource_name, path in self.at.items():
            if any(index < 0 for index in path):
                raise ValueError(
                    f"add_resource_transforms: `at` path for resource "
                    f"{resource_name!r} has a negative index: {path}"
                )
        for proto in self.transforms:
            if not proto.name:
                raise ValueError(
                    "add_resource_transforms: registry transforms must define "
                    "a non-empty name"
                )
        return self

Attributes

additions = PydanticField(..., description='Per-resource transform steps to append: ``{resource_name: [step_dict, ...]}``.', min_length=1) class-attribute instance-attribute

at = PydanticField(default_factory=dict, description='Per-resource pipeline level to append into: ``{resource_name: [step_index, ...]}``. Each index must address a ``descend`` step, descending one level per element; an omitted or empty path means the root level.') class-attribute instance-attribute

op = 'add_resource_transforms' class-attribute instance-attribute

transforms = PydanticField(default_factory=list, description='Named transforms to register in ``ingestion_model.transforms`` for steps that reference them via ``call.use``. A name already registered with a different body is an error at apply time.') class-attribute instance-attribute

AddResourcesOp

Bases: ConfigBaseModel

Introduce ingestion resources, in the shape the ingestion block accepts.

The unary way to grow ingestion_model: without it a change set could only ever rename or narrow the resources it started with, and the differ had to report an added resource as inexpressible.

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

class AddResourcesOp(ConfigBaseModel):
    """Introduce ingestion resources, in the shape the ingestion block accepts.

    The unary way to grow ``ingestion_model``: without it a change set could
    only ever rename or narrow the resources it started with, and the differ
    had to report an added resource as inexpressible.
    """

    op: Literal["add_resources"] = "add_resources"
    resources: list[ResourceConfig] = PydanticField(
        ...,
        description="Full resource definitions.",
        min_length=1,
    )
    transforms: list[ProtoTransform] = PydanticField(
        default_factory=list,
        description=(
            "Named transforms to register in ``ingestion_model.transforms`` for "
            "steps of the new resources that reference them via ``call.use``. "
            "Unioned by name exactly as ``add_resource_transforms`` does: an "
            "identical body already registered dedupes, a different one is an "
            "error at apply time."
        ),
    )

    @model_validator(mode="after")
    def _validate_unique_names(self) -> AddResourcesOp:
        names = [resource.name for resource in self.resources]
        if len(names) != len(set(names)):
            raise ValueError("add_resources entries must be unique by name")
        return self

Attributes

op = 'add_resources' class-attribute instance-attribute

resources = PydanticField(..., description='Full resource definitions.', min_length=1) class-attribute instance-attribute

transforms = PydanticField(default_factory=list, description='Named transforms to register in ``ingestion_model.transforms`` for steps of the new resources that reference them via ``call.use``. Unioned by name exactly as ``add_resource_transforms`` does: an identical body already registered dedupes, a different one is an error at apply time.') class-attribute instance-attribute

AddSecondaryIdentitiesOp

Bases: ConfigBaseModel

Declare alternate lookup keys on existing vertices.

Secondary identities are lookup-only: upserts keep using the primary identity. Each declared field-set automatically gains a non-unique index at :meth:Schema.finish_init, so this op is how an edge-only source gains a way to reference endpoints by a business key without touching the primary identity.

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

class AddSecondaryIdentitiesOp(ConfigBaseModel):
    """Declare alternate lookup keys on existing vertices.

    Secondary identities are lookup-only: upserts keep using the primary identity.
    Each declared field-set automatically gains a non-unique index at
    :meth:`Schema.finish_init`, so this op is how an edge-only source gains a way to
    reference endpoints by a business key without touching the primary identity.
    """

    op: Literal["add_secondary_identities"] = "add_secondary_identities"
    additions: dict[str, list[SecondaryIdentity]] = PydanticField(
        ...,
        description=(
            "Per-vertex secondary identities to declare: "
            "``{vertex_name: [{name, fields}, ...]}``. A bare field list is accepted "
            "for each entry and auto-named."
        ),
        min_length=1,
    )

Attributes

additions = PydanticField(..., description='Per-vertex secondary identities to declare: ``{vertex_name: [{name, fields}, ...]}``. A bare field list is accepted for each entry and auto-named.', min_length=1) class-attribute instance-attribute

op = 'add_secondary_identities' class-attribute instance-attribute

AddVertexIndexesOp

Bases: ConfigBaseModel

Author secondary indexes on vertices in the database profile.

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

class AddVertexIndexesOp(ConfigBaseModel):
    """Author secondary indexes on vertices in the database profile."""

    op: Literal["add_vertex_indexes"] = "add_vertex_indexes"
    indexes: dict[str, Annotated[list[Index], PydanticField(min_length=1)]] = (
        PydanticField(
            ...,
            description="``{vertex_name: [Index, ...]}``.",
            min_length=1,
        )
    )

Attributes

indexes = PydanticField(..., description='``{vertex_name: [Index, ...]}``.', min_length=1) class-attribute instance-attribute

op = 'add_vertex_indexes' class-attribute instance-attribute

AddVertexPropertiesOp

Bases: ConfigBaseModel

Add vertex properties to existing logical vertex types.

An entry may be a bare name or a full :class:~graflo.architecture.schema.vertex.Field. Bare names were the original shape and still mean what they meant -- an untyped property -- but they cannot express a property that arrives with a type and a grounding, which is what any op stream that adds a measured or temporal property has to say in one replayable step.

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

class AddVertexPropertiesOp(ConfigBaseModel):
    """Add vertex properties to existing logical vertex types.

    An entry may be a bare name or a full :class:`~graflo.architecture.schema.vertex.Field`.
    Bare names were the original shape and still mean what they meant -- an
    untyped property -- but they cannot express a property that arrives with a
    type and a grounding, which is what any op stream that adds a *measured* or
    *temporal* property has to say in one replayable step.
    """

    op: Literal["add_vertex_properties"] = "add_vertex_properties"
    additions: dict[str, list[str | Field]] = PydanticField(
        ...,
        description=(
            "Per-vertex property additions: ``{vertex_name: [field_name | Field, ...]}``."
        ),
        min_length=1,
    )

    def field_names(self, vertex: str) -> list[str]:
        """The names added to *vertex*, whichever shape they were written in."""
        return [
            entry if isinstance(entry, str) else entry.name
            for entry in self.additions.get(vertex, [])
        ]

Attributes

additions = PydanticField(..., description='Per-vertex property additions: ``{vertex_name: [field_name | Field, ...]}``.', min_length=1) class-attribute instance-attribute

op = 'add_vertex_properties' class-attribute instance-attribute

Methods:

field_names(vertex)

The names added to vertex, whichever shape they were written in.

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

def field_names(self, vertex: str) -> list[str]:
    """The names added to *vertex*, whichever shape they were written in."""
    return [
        entry if isinstance(entry, str) else entry.name
        for entry in self.additions.get(vertex, [])
    ]

AddVerticesOp

Bases: ConfigBaseModel

Introduce new logical vertex types.

The unary counterpart to what :class:MergeManifestsOp can only do binarily. A replayable change set that cannot introduce a type could only ever describe a shrinking graph, which is why this exists alongside remove_vertices.

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

class AddVerticesOp(ConfigBaseModel):
    """Introduce new logical vertex types.

    The unary counterpart to what :class:`MergeManifestsOp` can only do binarily.
    A replayable change set that cannot introduce a type could only ever describe a
    shrinking graph, which is why this exists alongside ``remove_vertices``.
    """

    op: Literal["add_vertices"] = "add_vertices"
    vertices: list[Vertex] = PydanticField(
        ...,
        description="Full vertex definitions, in the shape the schema block accepts.",
        min_length=1,
    )

    @model_validator(mode="after")
    def _validate_unique_names(self) -> AddVerticesOp:
        names = [vertex.name for vertex in self.vertices]
        if len(names) != len(set(names)):
            raise ValueError("add_vertices entries must be unique by name")
        return self

Attributes

op = 'add_vertices' class-attribute instance-attribute

vertices = PydanticField(..., description='Full vertex definitions, in the shape the schema block accepts.', min_length=1) class-attribute instance-attribute

AlignmentAttribute

Bases: ConfigBaseModel

One aligned canonical attribute; list position = funnel priority.

Each entry lowers to one :class:~graflo.architecture.schema.identity_funnel.IdentityBranch over name, so the list order is the funnel order. name is what the derived attribute is called — nothing collapses onto it, which is why it is not into; into is accepted as a legacy alias.

sources is keyed by resource because derivation inputs are that resource's raw column names. An entry takes one of three shapes:

• a single :class:DerivationSpec — the resource produces one member of the cluster, or all its members share the key column;
• a list of specs — the resource produces several members, each with its own key column, and at most one spec yields a value for any document (the others read an empty column). Lowers to scratch fields plus a coalesce_fields step;
• a dict keyed by member class — the resource produces several members and which one a document is must decide the derivation (the members share a column, or each carries its own marker). The lowering asks the side manifest how the resource produces each member and guards the step accordingly (when on the router's discriminator, or nothing for a plain vertex step). A member is keyed by its own name on its side or, through merge_manifests, by its canonical name;
• a :class:SharedDerivation — the same dict, spelled once: one call shared by the listed members, with only the parameters that differ.

Behind a vertex_router the first two shapes are guarded as well: the lowering reads which discriminator values route onto the class and puts them in when, so the steps run for no other class's documents.

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

class AlignmentAttribute(ConfigBaseModel):
    """One aligned canonical attribute; list position = funnel priority.

    Each entry lowers to one
    :class:`~graflo.architecture.schema.identity_funnel.IdentityBranch` over
    ``name``, so the list order *is* the funnel order. ``name`` is what the
    derived attribute is called — nothing collapses onto it, which is why it is
    not ``into``; ``into`` is accepted as a legacy alias.

    ``sources`` is keyed by resource because derivation inputs are that
    resource's raw column names. An entry takes one of three shapes:

    * a single :class:`DerivationSpec` — the resource produces one member of
      the cluster, or all its members share the key column;
    * a **list** of specs — the resource produces several members, each with
      its **own key column**, and at most one spec yields a value for any
      document (the others read an empty column). Lowers to scratch fields
      plus a ``coalesce_fields`` step;
    * a **dict keyed by member class** — the resource produces several members
      and which one a document *is* must decide the derivation (the members
      share a column, or each carries its own marker). The lowering asks the
      side manifest how the resource produces each member and guards the step
      accordingly (``when`` on the router's discriminator, or nothing for a
      plain ``vertex`` step). A member is keyed by its own name on its
      side or, through ``merge_manifests``, by its canonical name;
    * a :class:`SharedDerivation` — the same dict, spelled once: one call
      shared by the listed members, with only the parameters that differ.

    Behind a ``vertex_router`` the first two shapes are guarded as well: the
    lowering reads which discriminator values route onto the class and puts
    them in ``when``, so the steps run for no other class's documents.
    """

    name: str = PydanticField(
        ...,
        validation_alias=AliasChoices("name", "into"),
        description=(
            "Canonical attribute name on the class; funnel branch id. ``into`` "
            "is accepted as a legacy alias."
        ),
    )
    sources: dict[
        str,
        DerivationSpec
        | SharedDerivation
        | list[DerivationSpec]
        | dict[str, DerivationSpec],
    ] = PydanticField(
        ...,
        min_length=1,
        description=(
            "Per-resource derivation: ``{resource: spec}``; ``{resource: "
            "[spec, ...]}`` when the members the resource produces carry "
            "different key columns; ``{resource: {member_class: spec}}`` when "
            "the member a document is must decide the derivation, or a "
            "``SharedDerivation`` spelling that dict once."
        ),
    )

    def _by_member(self, resource: str) -> dict[str, DerivationSpec] | None:
        spec = self.sources.get(resource)
        if isinstance(spec, SharedDerivation):
            return spec.expand()
        return spec if isinstance(spec, dict) else None

    def specs_for(self, resource: str) -> list[DerivationSpec]:
        """Derivations *resource* contributes to this attribute, in order."""
        spec = self.sources.get(resource)
        if spec is None:
            return []
        if isinstance(spec, DerivationSpec):
            return [spec]
        if isinstance(spec, list):
            return list(spec)
        by_member = spec.expand() if isinstance(spec, SharedDerivation) else spec
        return list(by_member.values())

    def members_for(self, resource: str) -> list[str] | None:
        """Member classes keying *resource*'s specs, or ``None`` if unkeyed."""
        by_member = self._by_member(resource)
        return list(by_member) if by_member is not None else None

Attributes

name = PydanticField(..., validation_alias=AliasChoices('name', 'into'), description='Canonical attribute name on the class; funnel branch id. ``into`` is accepted as a legacy alias.') class-attribute instance-attribute

sources = PydanticField(..., min_length=1, description='Per-resource derivation: ``{resource: spec}``; ``{resource: [spec, ...]}`` when the members the resource produces carry different key columns; ``{resource: {member_class: spec}}`` when the member a document is must decide the derivation, or a ``SharedDerivation`` spelling that dict once.') class-attribute instance-attribute

Methods:

members_for(resource)

Member classes keying resource's specs, or None if unkeyed.

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

def members_for(self, resource: str) -> list[str] | None:
    """Member classes keying *resource*'s specs, or ``None`` if unkeyed."""
    by_member = self._by_member(resource)
    return list(by_member) if by_member is not None else None

specs_for(resource)

Derivations resource contributes to this attribute, in order.

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

def specs_for(self, resource: str) -> list[DerivationSpec]:
    """Derivations *resource* contributes to this attribute, in order."""
    spec = self.sources.get(resource)
    if spec is None:
        return []
    if isinstance(spec, DerivationSpec):
        return [spec]
    if isinstance(spec, list):
        return list(spec)
    by_member = spec.expand() if isinstance(spec, SharedDerivation) else spec
    return list(by_member.values())

AssignedIdentityTarget

Bases: ConfigBaseModel

Target an assigned identity: an intentional UUID primary key.

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

class AssignedIdentityTarget(ConfigBaseModel):
    """Target an assigned identity: an intentional UUID primary key."""

    mode: Literal["assigned"] = "assigned"

Attributes

mode = 'assigned' class-attribute instance-attribute

BlankIdentityTarget

Bases: ConfigBaseModel

Target a blank identity: an auto-generated placeholder ID.

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

class BlankIdentityTarget(ConfigBaseModel):
    """Target a blank identity: an auto-generated placeholder ID."""

    mode: Literal["blank"] = "blank"

Attributes

mode = 'blank' class-attribute instance-attribute

CanonicalMap

Bases: ConfigBaseModel

Declared translation of a source vocabulary into canonical names.

A partial function on names, identity where unmapped: vertices maps source class names to canonical class names, relations does the same for relation names, and properties maps, per source class name, source attribute names to canonical attribute names — including for classes whose name does not change. Two sources sharing a target is a merge and must be acknowledged with allow_merges.

It is a vocabulary, so it is idempotent: a canonical name is a fixed point that no entry maps away from. A chain ({X: Z, Z: Q}) or a swap is refused at construction — that shape is a relabel, which :class:CanonicalizeOp expresses directly. The rule is what lets two maps, or a map and an equivalence, be checked for agreement without asking in which order they were written.

Used on its own through :func:~graflo.architecture.evolution.canonical.canonical_map_to_ops, and on :attr:MergeManifestsOp.canonical_maps where it names the merged classes and is checked against the declared equivalences.

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

class CanonicalMap(ConfigBaseModel):
    """Declared translation of a source vocabulary into canonical names.

    A partial function on names, identity where unmapped: ``vertices`` maps
    source class names to canonical class names, ``relations`` does the same
    for relation names, and ``properties`` maps, per *source* class name,
    source attribute names to canonical attribute names — including for
    classes whose name does not change. Two sources sharing a target is a
    merge and must be acknowledged with ``allow_merges``.

    It is a **vocabulary**, so it is idempotent: a canonical name is a fixed
    point that no entry maps away from. A chain (``{X: Z, Z: Q}``) or a swap
    is refused at construction — that shape is a relabel, which
    :class:`CanonicalizeOp` expresses directly. The rule is what lets two
    maps, or a map and an equivalence, be checked for agreement without
    asking in which order they were written.

    Used on its own through :func:`~graflo.architecture.evolution.canonical.canonical_map_to_ops`,
    and on :attr:`MergeManifestsOp.canonical_maps` where it names the
    merged classes and is checked against the declared equivalences.
    """

    vertices: dict[str, str] = PydanticField(
        default_factory=dict,
        description="Class rename map: ``{source_class: canonical_class}``.",
    )
    properties: dict[str, dict[str, str]] = PydanticField(
        default_factory=dict,
        description=(
            "Per-source-class attribute rename map: "
            "``{source_class: {source_attr: canonical_attr}}``."
        ),
    )
    relations: dict[str, str] = PydanticField(
        default_factory=dict,
        description="Relation rename map: ``{source_relation: canonical_relation}``.",
    )
    allow_merges: bool = PydanticField(
        default=False,
        description=(
            "Accept a non-injective ``vertices`` / ``relations`` map. Two "
            "sources sharing a canonical target is a *merge*, not a rename; "
            "it must be a stated intent because merging fuses entities and "
            "can create self-relations."
        ),
    )
    allow_dangling_entries: bool = PydanticField(
        default=False,
        description=(
            "Accept entries that name nothing in the manifest the map is "
            "applied to, dropping and logging each one. A shared vocabulary "
            "map is legitimately broader than any single manifest. Off by "
            "default, because a misspelt class has exactly the same shape, "
            "and dropping it silently narrows the rename to less than the "
            "author asked for."
        ),
    )

    @model_validator(mode="after")
    def _validate_maps(self) -> CanonicalMap:
        if not self.allow_merges:
            # Identity entries (source == target) are excluded: a lowered
            # cluster map deliberately carries one for every member, including
            # the merged name itself, to declare it a member of its own group
            # -- that self entry must not read as a collision here. The op the
            # map lowers to counts it, which is where the merge is acknowledged.
            validate_rename_map_is_injective(
                {s: t for s, t in self.vertices.items() if s != t},
                kind="canonical vertex",
                merge_hint="CanonicalMap(allow_merges=True)",
            )
            validate_rename_map_is_injective(
                {s: t for s, t in self.relations.items() if s != t},
                kind="canonical relation",
                merge_hint="CanonicalMap(allow_merges=True)",
            )
        validate_vocabulary_is_idempotent(self.vertices, kind="canonical vertex")
        validate_vocabulary_is_idempotent(self.relations, kind="canonical relation")
        for source_class, attr_map in self.properties.items():
            validate_rename_map_is_injective(
                attr_map,
                kind=f"canonical property (class {source_class!r})",
                merge_hint="a transform that combines the fields upstream",
            )
            validate_vocabulary_is_idempotent(
                attr_map, kind=f"canonical property (class {source_class!r})"
            )
        return self

    def canonical_class(self, source_class: str) -> str:
        """Canonical name of *source_class* (itself when unmapped)."""
        return self.vertices.get(source_class, source_class)

    def canonical_relation(self, source_relation: str) -> str:
        """Canonical name of *source_relation* (itself when unmapped)."""
        return self.relations.get(source_relation, source_relation)

    @property
    def vertex_targets(self) -> set[str]:
        """Canonical class names this map establishes (targets of a real rename)."""
        return {t for s, t in self.vertices.items() if s != t}

    @property
    def relation_targets(self) -> set[str]:
        """Canonical relation names this map establishes."""
        return {t for s, t in self.relations.items() if s != t}

    def canonical_property_names(self, canonical_class: str) -> set[str]:
        """Canonical attribute names the map establishes on *canonical_class*."""
        names: set[str] = set()
        for source_class, attr_map in self.properties.items():
            if self.canonical_class(source_class) == canonical_class:
                names.update(new for old, new in attr_map.items() if old != new)
        return names

Attributes

allow_dangling_entries = PydanticField(default=False, description='Accept entries that name nothing in the manifest the map is applied to, dropping and logging each one. A shared vocabulary map is legitimately broader than any single manifest. Off by default, because a misspelt class has exactly the same shape, and dropping it silently narrows the rename to less than the author asked for.') class-attribute instance-attribute

allow_merges = PydanticField(default=False, description='Accept a non-injective ``vertices`` / ``relations`` map. Two sources sharing a canonical target is a *merge*, not a rename; it must be a stated intent because merging fuses entities and can create self-relations.') class-attribute instance-attribute

properties = PydanticField(default_factory=dict, description='Per-source-class attribute rename map: ``{source_class: {source_attr: canonical_attr}}``.') class-attribute instance-attribute

relation_targets property

Canonical relation names this map establishes.

relations = PydanticField(default_factory=dict, description='Relation rename map: ``{source_relation: canonical_relation}``.') class-attribute instance-attribute

vertex_targets property

Canonical class names this map establishes (targets of a real rename).

vertices = PydanticField(default_factory=dict, description='Class rename map: ``{source_class: canonical_class}``.') class-attribute instance-attribute

Methods:

canonical_class(source_class)

Canonical name of source_class (itself when unmapped).

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

def canonical_class(self, source_class: str) -> str:
    """Canonical name of *source_class* (itself when unmapped)."""
    return self.vertices.get(source_class, source_class)

canonical_property_names(canonical_class)

Canonical attribute names the map establishes on canonical_class.

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

def canonical_property_names(self, canonical_class: str) -> set[str]:
    """Canonical attribute names the map establishes on *canonical_class*."""
    names: set[str] = set()
    for source_class, attr_map in self.properties.items():
        if self.canonical_class(source_class) == canonical_class:
            names.update(new for old, new in attr_map.items() if old != new)
    return names

canonical_relation(source_relation)

Canonical name of source_relation (itself when unmapped).

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

def canonical_relation(self, source_relation: str) -> str:
    """Canonical name of *source_relation* (itself when unmapped)."""
    return self.relations.get(source_relation, source_relation)

CanonicalizeOp

Bases: ConfigBaseModel

Relabel classes, attributes and relations by one vocabulary map, in one step.

The map is a partial function on names — identity where unmapped — applied simultaneously over the original schema, so a chain ({X: Z, Z: Q}) and a swap resolve without an intermediate state, and the fibers of the map are exactly the groups that merge. A target that already exists and does not move must be declared a member of its own group with a self entry (Company: Company); otherwise the op refuses rather than merging into it silently. properties is keyed by the source class name and is applied before the class relabel.

This is the single lowering of a :class:~graflo.architecture.evolution.canonical.CanonicalMap, and the per-side step of :func:~graflo.architecture.evolution.merge.merge_manifests.

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

class CanonicalizeOp(ConfigBaseModel):
    """Relabel classes, attributes and relations by one vocabulary map, in one step.

    The map is a partial function on names — identity where unmapped — applied
    simultaneously over the original schema, so a chain (``{X: Z, Z: Q}``) and
    a swap resolve without an intermediate state, and the fibers of the map
    are exactly the groups that merge. A target that already exists and does
    not move must be declared a member of its own group with a self entry
    (``Company: Company``); otherwise the op refuses rather than merging into
    it silently. ``properties`` is keyed by the *source* class name and is
    applied before the class relabel.

    This is the single lowering of a
    :class:`~graflo.architecture.evolution.canonical.CanonicalMap`, and the
    per-side step of
    :func:`~graflo.architecture.evolution.merge.merge_manifests`.
    """

    op: Literal["canonicalize"] = "canonicalize"
    vertices: dict[str, str] = PydanticField(
        default_factory=dict,
        description="Class map: ``{source_class: canonical_class}``.",
    )
    properties: dict[str, dict[str, str]] = PydanticField(
        default_factory=dict,
        description=(
            "Per-source-class attribute map: "
            "``{source_class: {source_attr: canonical_attr}}``."
        ),
    )
    relations: dict[str, str] = PydanticField(
        default_factory=dict,
        description="Relation map: ``{source_relation: canonical_relation}``.",
    )
    allow_merges: bool = PydanticField(
        default=False,
        description=(
            "Accept a group of more than one class or relation collapsing onto "
            "one target. A merge fuses entities and can create self-relations, "
            "so it is acknowledged here rather than inferred from the map."
        ),
    )
    allow_self_relations: bool = PydanticField(
        default=False,
        description=(
            "Accept a merge whose sources are connected by an edge that becomes "
            "a self-relation once both endpoints land on the same class."
        ),
    )
    allow_observation_fusion: bool = PydanticField(
        default=False,
        description=(
            "Accept a merge whose sources are produced in one accumulator slot "
            "(the same pipeline level and the same ``role``, or both bare), "
            "fusing those observations into one node."
        ),
    )

    @model_validator(mode="after")
    def _validate_map(self) -> CanonicalizeOp:
        validate_vocabulary_map(
            self.vertices,
            self.relations,
            self.properties,
            allow_merges=self.allow_merges,
            kind="canonicalize",
            merge_hint="set allow_merges=true",
        )
        return self

    @property
    def vertex_groups(self) -> dict[str, list[str]]:
        """``{target: [members]}`` over ``vertices``, self entries included."""
        return vocabulary_groups(self.vertices)

    @property
    def relation_groups(self) -> dict[str, list[str]]:
        """``{target: [members]}`` over ``relations``, self entries included."""
        return vocabulary_groups(self.relations)

    @property
    def merges(self) -> bool:
        """Whether any class or relation group has more than one member."""
        return any(
            len(members) > 1
            for groups in (self.vertex_groups, self.relation_groups)
            for members in groups.values()
        )

Attributes

allow_merges = PydanticField(default=False, description='Accept a group of more than one class or relation collapsing onto one target. A merge fuses entities and can create self-relations, so it is acknowledged here rather than inferred from the map.') class-attribute instance-attribute

allow_observation_fusion = PydanticField(default=False, description='Accept a merge whose sources are produced in one accumulator slot (the same pipeline level and the same ``role``, or both bare), fusing those observations into one node.') class-attribute instance-attribute

allow_self_relations = PydanticField(default=False, description='Accept a merge whose sources are connected by an edge that becomes a self-relation once both endpoints land on the same class.') class-attribute instance-attribute

merges property

Whether any class or relation group has more than one member.

op = 'canonicalize' class-attribute instance-attribute

properties = PydanticField(default_factory=dict, description='Per-source-class attribute map: ``{source_class: {source_attr: canonical_attr}}``.') class-attribute instance-attribute

relation_groups property

{target: [members]} over relations, self entries included.

relations = PydanticField(default_factory=dict, description='Relation map: ``{source_relation: canonical_relation}``.') class-attribute instance-attribute

vertex_groups property

{target: [members]} over vertices, self entries included.

vertices = PydanticField(default_factory=dict, description='Class map: ``{source_class: canonical_class}``.') class-attribute instance-attribute

ChangeFieldTypesOp

Bases: ConfigBaseModel

Set the logical type of existing vertex or edge properties.

Makes the differ's CHANGE_VERTEX_FIELD_TYPE / CHANGE_EDGE_FIELD_TYPE authorable. Targets are validated against the profile's db_flavor so an unsupported type fails here rather than at define time.

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

class ChangeFieldTypesOp(ConfigBaseModel):
    """Set the logical type of existing vertex or edge properties.

    Makes the differ's ``CHANGE_VERTEX_FIELD_TYPE`` / ``CHANGE_EDGE_FIELD_TYPE``
    authorable. Targets are validated against the profile's ``db_flavor`` so an
    unsupported type fails here rather than at define time.
    """

    op: Literal["change_field_types"] = "change_field_types"
    vertices: dict[str, dict[str, FieldTypeSpec]] = PydanticField(
        default_factory=dict,
        description="``{vertex_name: {field_name: {type, item_type}}}``.",
    )
    edges: dict[str, dict[str, FieldTypeSpec]] = PydanticField(
        default_factory=dict,
        description="``{relation_name: {field_name: {type, item_type}}}``.",
    )

    @model_validator(mode="after")
    def _require_a_target(self) -> ChangeFieldTypesOp:
        if not self.vertices and not self.edges:
            raise ValueError(
                "change_field_types requires at least one of vertices or edges"
            )
        return self

Attributes

edges = PydanticField(default_factory=dict, description='``{relation_name: {field_name: {type, item_type}}}``.') class-attribute instance-attribute

op = 'change_field_types' class-attribute instance-attribute

vertices = PydanticField(default_factory=dict, description='``{vertex_name: {field_name: {type, item_type}}}``.') class-attribute instance-attribute

DeclareEdgeInversesOp

Bases: ConfigBaseModel

Declare inverse pairs and symmetric relations in edge_config.

Purely logical: it records how relation names read one fact from its two endpoints and creates nothing. inverses pairs two distinct names; a pair is unordered, so {a: b}, {b: a} and {a: b, b: a} declare the same thing. symmetric names relations that are their own inverse. Together they must give every relation at most one inverse (no a: b with b: c), in the op and against what is already declared.

Realize a pair with :class:AddInverseEdgesOp (explicit, portable inverse edges) or :class:SetNativeInversesOp (TigerGraph maintains the pair) -- never both for one relation. A symmetric relation is realized by its edges being undirected (:class:SetEdgeDirectedOp), which the schema requires.

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

class DeclareEdgeInversesOp(ConfigBaseModel):
    """Declare inverse pairs and symmetric relations in ``edge_config``.

    Purely logical: it records how relation names read one fact from its two
    endpoints and creates nothing. ``inverses`` pairs two distinct names; a pair
    is unordered, so ``{a: b}``, ``{b: a}`` and ``{a: b, b: a}`` declare the same
    thing. ``symmetric`` names relations that are their own inverse. Together
    they must give every relation at most one inverse (no ``a: b`` with
    ``b: c``), in the op and against what is already declared.

    Realize a pair with :class:`AddInverseEdgesOp` (explicit, portable inverse
    edges) or :class:`SetNativeInversesOp` (TigerGraph maintains the pair) --
    never both for one relation. A symmetric relation is realized by its edges
    being undirected (:class:`SetEdgeDirectedOp`), which the schema requires.
    """

    op: Literal["declare_edge_inverses"] = "declare_edge_inverses"
    inverses: dict[str, str] = PydanticField(
        default_factory=dict,
        description="Pairs to declare: ``{relation: inverse_relation}``, either order.",
    )
    symmetric: list[str] = PydanticField(
        default_factory=list,
        description="Relations to declare as their own inverse.",
    )

    @model_validator(mode="after")
    def _validate_table(self) -> DeclareEdgeInversesOp:
        if not self.inverses and not self.symmetric:
            raise ValueError(
                "declare_edge_inverses: nothing to declare; give inverses or symmetric"
            )
        normalize_inverse_table(
            self.inverses.items(), self.symmetric, kind="declare_edge_inverses"
        )
        return self

Attributes

inverses = PydanticField(default_factory=dict, description='Pairs to declare: ``{relation: inverse_relation}``, either order.') class-attribute instance-attribute

op = 'declare_edge_inverses' class-attribute instance-attribute

symmetric = PydanticField(default_factory=list, description='Relations to declare as their own inverse.') class-attribute instance-attribute

DerivationSpec

Bases: ConfigBaseModel

How one resource derives a canonical attribute from its raw doc fields.

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

class DerivationSpec(ConfigBaseModel):
    """How one resource derives a canonical attribute from its raw doc fields."""

    input: list[str] = PydanticField(
        ...,
        min_length=1,
        description=(
            "RAW source-doc field names fed to the function, in order. "
            "Documents keep their original keys after property renames, so "
            "canonical property names are usually wrong here."
        ),
    )
    module: str = PydanticField(
        default="graflo.util.transform",
        description="Module holding the derivation function.",
    )
    foo: str = PydanticField(
        default="gated_normalized_key",
        description="Function name; called as ``foo(*values, **params)``.",
    )
    params: dict[str, Any] = PydanticField(
        default_factory=dict,
        description="Keyword parameters for the function (gate prefix, ...).",
    )

Attributes

foo = PydanticField(default='gated_normalized_key', description='Function name; called as ``foo(*values, **params)``.') class-attribute instance-attribute

input = PydanticField(..., min_length=1, description='RAW source-doc field names fed to the function, in order. Documents keep their original keys after property renames, so canonical property names are usually wrong here.') class-attribute instance-attribute

module = PydanticField(default='graflo.util.transform', description='Module holding the derivation function.') class-attribute instance-attribute

params = PydanticField(default_factory=dict, description='Keyword parameters for the function (gate prefix, ...).') class-attribute instance-attribute

EdgeFieldSemanticsTarget

Bases: ConfigBaseModel

One property of one edge triple, and the grounding to put on it.

Edge properties carry the same FieldSemantics as vertex properties -- a since on an edge has a unit as much as a temperature on a vertex does -- but until this target existed no op could reach them.

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

class EdgeFieldSemanticsTarget(ConfigBaseModel):
    """One property of one edge triple, and the grounding to put on it.

    Edge properties carry the same ``FieldSemantics`` as vertex properties --
    a ``since`` on an edge has a unit as much as a ``temperature`` on a vertex
    does -- but until this target existed no op could reach them.
    """

    source: str = PydanticField(..., description="Source vertex type name.")
    target: str = PydanticField(..., description="Target vertex type name.")
    relation: str | None = PydanticField(
        default=None,
        description="Relation name; ``None`` matches the edge with no relation set.",
    )
    field: str = PydanticField(..., description="Property name on that edge.")
    semantics: FieldSemantics | None = PydanticField(
        default=None,
        description="Grounding for the property; ``None`` clears it.",
    )

    def edge_id(self) -> tuple[str, str, str | None]:
        return self.source, self.target, self.relation

    def key(self) -> tuple[Any, ...]:
        return ("edge", self.source, self.target, self.relation, self.field)

Attributes

field = PydanticField(..., description='Property name on that edge.') class-attribute instance-attribute

relation = PydanticField(default=None, description='Relation name; ``None`` matches the edge with no relation set.') class-attribute instance-attribute

semantics = PydanticField(default=None, description='Grounding for the property; ``None`` clears it.') class-attribute instance-attribute

source = PydanticField(..., description='Source vertex type name.') class-attribute instance-attribute

target = PydanticField(..., description='Target vertex type name.') class-attribute instance-attribute

Methods:

edge_id()

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

def edge_id(self) -> tuple[str, str, str | None]:
    return self.source, self.target, self.relation

key()

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

def key(self) -> tuple[Any, ...]:
    return ("edge", self.source, self.target, self.relation, self.field)

EdgeIdentitiesEntry

Bases: ConfigBaseModel

New uniqueness keys for one edge triple.

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

class EdgeIdentitiesEntry(ConfigBaseModel):
    """New uniqueness keys for one edge triple."""

    source: str = PydanticField(..., description="Source vertex type name.")
    target: str = PydanticField(..., description="Target vertex type name.")
    relation: str | None = PydanticField(
        default=None,
        description="Relation name; ``None`` matches the edge with no relation set.",
    )
    identities: list[list[str]] = PydanticField(
        ...,
        description=(
            "Replacement uniqueness keys. Each key lists fields that, together with "
            "the resolved endpoints, must be unique; the ``source`` / ``target`` "
            "tokens stand for the endpoints themselves. An empty list clears them."
        ),
    )

    def edge_id(self) -> tuple[str, str, str | None]:
        return self.source, self.target, self.relation

Attributes

identities = PydanticField(..., description='Replacement uniqueness keys. Each key lists fields that, together with the resolved endpoints, must be unique; the ``source`` / ``target`` tokens stand for the endpoints themselves. An empty list clears them.') class-attribute instance-attribute

relation = PydanticField(default=None, description='Relation name; ``None`` matches the edge with no relation set.') class-attribute instance-attribute

source = PydanticField(..., description='Source vertex type name.') class-attribute instance-attribute

target = PydanticField(..., description='Target vertex type name.') class-attribute instance-attribute

Methods:

edge_id()

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

def edge_id(self) -> tuple[str, str, str | None]:
    return self.source, self.target, self.relation

EdgeIndexEntry

Bases: ConfigBaseModel

Indexes for one edge physical spec.

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

class EdgeIndexEntry(ConfigBaseModel):
    """Indexes for one edge physical spec."""

    source: str = PydanticField(..., description="Source vertex type name.")
    target: str = PydanticField(..., description="Target vertex type name.")
    relation: str | None = PydanticField(
        default=None,
        description="Relation name; ``None`` matches the edge with no relation set.",
    )
    purpose: str | None = PydanticField(
        default=None,
        description="Physical variant purpose; ``None`` addresses the base spec.",
    )
    indexes: list[Index] = PydanticField(
        default_factory=list,
        description="Indexes to add to this spec.",
    )
    fields: list[list[str]] = PydanticField(
        default_factory=list,
        description="Field lists identifying indexes to remove from this spec.",
    )

    def physical_key(self) -> tuple[str, str, str | None, str | None]:
        return self.source, self.target, self.relation, self.purpose

Attributes

fields = PydanticField(default_factory=list, description='Field lists identifying indexes to remove from this spec.') class-attribute instance-attribute

indexes = PydanticField(default_factory=list, description='Indexes to add to this spec.') class-attribute instance-attribute

purpose = PydanticField(default=None, description='Physical variant purpose; ``None`` addresses the base spec.') class-attribute instance-attribute

relation = PydanticField(default=None, description='Relation name; ``None`` matches the edge with no relation set.') class-attribute instance-attribute

source = PydanticField(..., description='Source vertex type name.') class-attribute instance-attribute

target = PydanticField(..., description='Target vertex type name.') class-attribute instance-attribute

Methods:

physical_key()

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

def physical_key(self) -> tuple[str, str, str | None, str | None]:
    return self.source, self.target, self.relation, self.purpose

EdgeRetargetEntry

Bases: ConfigBaseModel

Repoint one edge triple at a different source and/or target vertex type.

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

class EdgeRetargetEntry(ConfigBaseModel):
    """Repoint one edge triple at a different source and/or target vertex type."""

    source: str = PydanticField(..., description="Current source vertex type name.")
    target: str = PydanticField(..., description="Current target vertex type name.")
    relation: str | None = PydanticField(
        default=None,
        description="Relation name; ``None`` matches the edge with no relation set.",
    )
    new_source: str | None = PydanticField(
        default=None,
        description="Replacement source vertex type; omit to keep the current one.",
    )
    new_target: str | None = PydanticField(
        default=None,
        description="Replacement target vertex type; omit to keep the current one.",
    )

    def edge_id(self) -> tuple[str, str, str | None]:
        return self.source, self.target, self.relation

    def retargeted_edge_id(self) -> tuple[str, str, str | None]:
        return (
            self.new_source or self.source,
            self.new_target or self.target,
            self.relation,
        )

    @model_validator(mode="after")
    def _require_a_change(self) -> EdgeRetargetEntry:
        if self.new_source is None and self.new_target is None:
            raise ValueError(
                "retarget_edges requires at least one of new_source or new_target"
            )
        if self.retargeted_edge_id() == self.edge_id():
            raise ValueError(
                f"retarget_edges: edge {self.edge_id()} would not change endpoints"
            )
        return self

Attributes

new_source = PydanticField(default=None, description='Replacement source vertex type; omit to keep the current one.') class-attribute instance-attribute

new_target = PydanticField(default=None, description='Replacement target vertex type; omit to keep the current one.') class-attribute instance-attribute

relation = PydanticField(default=None, description='Relation name; ``None`` matches the edge with no relation set.') class-attribute instance-attribute

source = PydanticField(..., description='Current source vertex type name.') class-attribute instance-attribute

target = PydanticField(..., description='Current target vertex type name.') class-attribute instance-attribute

Methods:

edge_id()

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

def edge_id(self) -> tuple[str, str, str | None]:
    return self.source, self.target, self.relation

retargeted_edge_id()

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

def retargeted_edge_id(self) -> tuple[str, str, str | None]:
    return (
        self.new_source or self.source,
        self.new_target or self.target,
        self.relation,
    )

EdgeSelector

Bases: ConfigBaseModel

Schema edge triple selector matching :data:~graflo.architecture.graph_types.EdgeId.

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

class EdgeSelector(ConfigBaseModel):
    """Schema edge triple selector matching :data:`~graflo.architecture.graph_types.EdgeId`."""

    source: str = PydanticField(..., description="Source vertex type name.")
    target: str = PydanticField(..., description="Target vertex type name.")
    relation: str | None = PydanticField(
        default=None,
        description="Relation name; ``None`` matches edges with no relation set.",
    )

    def edge_id(self) -> tuple[str, str, str | None]:
        return self.source, self.target, self.relation

Attributes

relation = PydanticField(default=None, description='Relation name; ``None`` matches edges with no relation set.') class-attribute instance-attribute

source = PydanticField(..., description='Source vertex type name.') class-attribute instance-attribute

target = PydanticField(..., description='Target vertex type name.') class-attribute instance-attribute

Methods:

edge_id()

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

def edge_id(self) -> tuple[str, str, str | None]:
    return self.source, self.target, self.relation

EnsureExtractedFields

Bases: ConfigBaseModel

Fields that must survive extraction for one vertex type at one level.

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

class EnsureExtractedFields(ConfigBaseModel):
    """Fields that must survive extraction for one vertex type at one level."""

    vertex: str = PydanticField(
        ...,
        description="The vertex type whose extraction must keep ``fields``.",
    )
    fields: list[str] = PydanticField(
        ...,
        min_length=1,
        description="Property names that must reach the extracted vertex document.",
    )
    at: list[int] = PydanticField(
        default_factory=list,
        description=(
            "Pipeline level holding the producing step, as ``descend`` step "
            "indices. Empty means the root level."
        ),
    )

Attributes

at = PydanticField(default_factory=list, description='Pipeline level holding the producing step, as ``descend`` step indices. Empty means the root level.') class-attribute instance-attribute

fields = PydanticField(..., min_length=1, description='Property names that must reach the extracted vertex document.') class-attribute instance-attribute

vertex = PydanticField(..., description='The vertex type whose extraction must keep ``fields``.') class-attribute instance-attribute

EnsureExtractedFieldsOp

Bases: ConfigBaseModel

Widen a producing step's projection so named fields are not dropped.

Needed because a vertex_router delivers differently from a vertex step. The router builds its child VertexActor at lindex.extend((role, 0)), where the transform buffer is empty, so derived fields reach the child only through the merged observation — that is, through passthrough or from. A plain vertex step instead reads the buffer directly, which bypasses keep_fields and extraction_scope entirely.

So on a router, extraction_scope: mapped_only or a keep_fields list that does not name the fields drops them silently. This op restores them: keep_fields gains the names, and under mapped_only the per-type vertex_from_map entry gains identity mappings — seeded from the router-level from when the entry does not exist yet, since creating it otherwise replaces the author's projection rather than extending it.

A plain vertex step, or a router that restricts nothing, is a no-op: the fields already survive.

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

class EnsureExtractedFieldsOp(ConfigBaseModel):
    """Widen a producing step's projection so named fields are not dropped.

    Needed because a ``vertex_router`` delivers differently from a ``vertex``
    step. The router builds its child ``VertexActor`` at
    ``lindex.extend((role, 0))``, where the transform buffer is empty, so
    derived fields reach the child only through the merged observation — that
    is, through passthrough or ``from``. A plain ``vertex`` step instead reads
    the buffer directly, which bypasses ``keep_fields`` and
    ``extraction_scope`` entirely.

    So on a router, ``extraction_scope: mapped_only`` or a ``keep_fields`` list
    that does not name the fields drops them silently. This op restores them:
    ``keep_fields`` gains the names, and under ``mapped_only`` the per-type
    ``vertex_from_map`` entry gains identity mappings — seeded from the
    router-level ``from`` when the entry does not exist yet, since creating it
    otherwise replaces the author's projection rather than extending it.

    A plain ``vertex`` step, or a router that restricts nothing, is a no-op:
    the fields already survive.
    """

    op: Literal["ensure_extracted_fields"] = "ensure_extracted_fields"
    additions: dict[str, list[EnsureExtractedFields]] = PydanticField(
        ...,
        description=(
            "Per-resource extraction guarantees: "
            "``{resource_name: [{vertex, fields, at}, ...]}``."
        ),
        min_length=1,
    )

    @model_validator(mode="after")
    def _validate_entries(self) -> EnsureExtractedFieldsOp:
        for resource_name, entries in self.additions.items():
            if not entries:
                raise ValueError(
                    f"ensure_extracted_fields: empty entry list for resource "
                    f"{resource_name!r}"
                )
            for entry in entries:
                if any(index < 0 for index in entry.at):
                    raise ValueError(
                        f"ensure_extracted_fields: `at` path for resource "
                        f"{resource_name!r} has a negative index: {entry.at}"
                    )
        return self

Attributes

additions = PydanticField(..., description='Per-resource extraction guarantees: ``{resource_name: [{vertex, fields, at}, ...]}``.', min_length=1) class-attribute instance-attribute

op = 'ensure_extracted_fields' class-attribute instance-attribute

FieldSemanticsTarget

Bases: ConfigBaseModel

One property of one vertex, and the grounding to put on it.

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

class FieldSemanticsTarget(ConfigBaseModel):
    """One property of one vertex, and the grounding to put on it."""

    vertex: str = PydanticField(..., description="Vertex type name.")
    field: str = PydanticField(..., description="Property name on that vertex.")
    semantics: FieldSemantics | None = PydanticField(
        default=None,
        description="Grounding for the property; ``None`` clears it.",
    )

    def key(self) -> tuple[Any, ...]:
        return ("vertex", self.vertex, self.field)

Attributes

field = PydanticField(..., description='Property name on that vertex.') class-attribute instance-attribute

semantics = PydanticField(default=None, description='Grounding for the property; ``None`` clears it.') class-attribute instance-attribute

vertex = PydanticField(..., description='Vertex type name.') class-attribute instance-attribute

Methods:

key()

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

def key(self) -> tuple[Any, ...]:
    return ("vertex", self.vertex, self.field)

FieldTypeSpec

Bases: ConfigBaseModel

Target logical type for one property.

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

class FieldTypeSpec(ConfigBaseModel):
    """Target logical type for one property."""

    type: FieldType | None = PydanticField(
        ...,
        description="New logical field type; ``None`` clears the declared type.",
    )
    item_type: FieldType | None = PydanticField(
        default=None,
        description="Element type, required when ``type`` is ``LIST``.",
    )

    @model_validator(mode="after")
    def _validate_item_type(self) -> FieldTypeSpec:
        if self.type == FieldType.LIST and self.item_type is None:
            raise ValueError("a LIST field type requires item_type")
        if self.type != FieldType.LIST and self.item_type is not None:
            raise ValueError("item_type is only meaningful for a LIST field type")
        return self

Attributes

item_type = PydanticField(default=None, description='Element type, required when ``type`` is ``LIST``.') class-attribute instance-attribute

type = PydanticField(..., description='New logical field type; ``None`` clears the declared type.') class-attribute instance-attribute

FunnelIdentityTarget

Bases: ConfigBaseModel

Target an identity funnel: ordered fallback branches digested into id.

The general form of :class:HashIdentityTarget — a flat hash key is a funnel with one branch. Both resolve to identity mode hash.

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

class FunnelIdentityTarget(ConfigBaseModel):
    """Target an identity funnel: ordered fallback branches digested into ``id``.

    The general form of :class:`HashIdentityTarget` — a flat hash key is a funnel
    with one branch. Both resolve to identity mode ``hash``.
    """

    mode: Literal["funnel"] = "funnel"
    funnel: IdentityFunnel = PydanticField(
        ...,
        description="Ordered fallback branches; the first complete one wins.",
    )

Attributes

funnel = PydanticField(..., description='Ordered fallback branches; the first complete one wins.') class-attribute instance-attribute

mode = 'funnel' class-attribute instance-attribute

HashIdentityTarget

Bases: ConfigBaseModel

Target a hash identity: a deterministic synthetic id digested from fields.

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

class HashIdentityTarget(ConfigBaseModel):
    """Target a hash identity: a deterministic synthetic ``id`` digested from fields."""

    mode: Literal["hash"] = "hash"
    hash_from: list[str] = PydanticField(
        ...,
        description="Source property names whose values are digested into ``id``.",
        min_length=1,
    )

Attributes

hash_from = PydanticField(..., description='Source property names whose values are digested into ``id``.', min_length=1) class-attribute instance-attribute

mode = 'hash' class-attribute instance-attribute

IdentityAlignment

Bases: ConfigBaseModel

Cross-source identity alignment for one canonical class.

attributes order is funnel priority: a record keys by the highest-priority aligned attribute it carries. Two records fuse when their strongest present attribute coincides — a match on a lower-priority attribute does NOT fuse records when one of them also carries a higher-priority one.

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

class IdentityAlignment(ConfigBaseModel):
    """Cross-source identity alignment for one canonical class.

    ``attributes`` order is funnel priority: a record keys by the highest-priority
    aligned attribute it carries. Two records fuse when their strongest
    present attribute coincides — a match on a lower-priority attribute does
    NOT fuse records when one of them also carries a higher-priority one.
    """

    vertex: str = PydanticField(
        ...,
        description="The canonical class whose identity is being aligned.",
    )
    attributes: list[AlignmentAttribute] = PydanticField(
        default_factory=list,
        validation_alias=AliasChoices("attributes", "rows"),
        description=(
            "Aligned canonical attributes, in priority order. ``rows`` is "
            "accepted as a legacy alias."
        ),
    )
    local_key: LocalKeySpec | None = PydanticField(
        default=None,
        description=(
            "Fallback identity for records carrying no aligned attribute. "
            "Without it such records get no identity and are dropped."
        ),
    )
    secondary_identities: dict[str, list[str]] = PydanticField(
        default_factory=dict,
        description=(
            "Retired side keys kept as lookup-only secondary identities: "
            "``{name: [field, ...]}``."
        ),
    )
    at: dict[str, list[int]] = PydanticField(
        default_factory=dict,
        description=(
            "Per-resource pipeline level to derive at, as ``descend`` step "
            "indices. Omitted resources resolve to the single level producing "
            "``vertex`` (for member-keyed sources: the level producing the "
            "member on its side); supply a path only when a resource produces "
            "it at more than one level."
        ),
    )

    @model_validator(mode="after")
    def _validate_shape(self) -> IdentityAlignment:
        if not self.attributes and self.local_key is None:
            raise ValueError(
                "IdentityAlignment requires at least one attribute or a local_key"
            )
        into_names = [attribute.name for attribute in self.attributes]
        if self.local_key is not None:
            into_names.append(self.local_key.name)
        duplicates = {n for n in into_names if into_names.count(n) > 1}
        if duplicates:
            raise ValueError(
                f"IdentityAlignment: duplicate target attributes {sorted(duplicates)}"
            )
        return self

Attributes

at = PydanticField(default_factory=dict, description='Per-resource pipeline level to derive at, as ``descend`` step indices. Omitted resources resolve to the single level producing ``vertex`` (for member-keyed sources: the level producing the member on its side); supply a path only when a resource produces it at more than one level.') class-attribute instance-attribute

attributes = PydanticField(default_factory=list, validation_alias=AliasChoices('attributes', 'rows'), description='Aligned canonical attributes, in priority order. ``rows`` is accepted as a legacy alias.') class-attribute instance-attribute

local_key = PydanticField(default=None, description='Fallback identity for records carrying no aligned attribute. Without it such records get no identity and are dropped.') class-attribute instance-attribute

secondary_identities = PydanticField(default_factory=dict, description='Retired side keys kept as lookup-only secondary identities: ``{name: [field, ...]}``.') class-attribute instance-attribute

vertex = PydanticField(..., description='The canonical class whose identity is being aligned.') class-attribute instance-attribute

IdentityReplacement

Bases: ConfigBaseModel

New identity policy for one vertex, plus what becomes of the old one.

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

class IdentityReplacement(ConfigBaseModel):
    """New identity policy for one vertex, plus what becomes of the old one."""

    to: IdentityTarget = PydanticField(
        ...,
        description="The identity policy this vertex should have after the op.",
    )
    retire: Literal["demote", "keep", "drop"] = PydanticField(
        default="demote",
        description=(
            "What happens to the old identity field-set. ``demote`` turns it into a "
            "secondary identity (lookup index follows automatically), ``keep`` leaves "
            "the fields as plain properties, ``drop`` removes them. Demotion is "
            "downgraded to ``keep`` when the old identity was synthetic (hash / "
            "assigned / blank) or already equals the new one."
        ),
    )
    retire_as: str | None = PydanticField(
        default=None,
        description=(
            "Name for the demoted secondary identity. Defaults to "
            "``retired_identity``. Only meaningful with ``retire: demote``."
        ),
    )
    endpoints: Literal["follow_new", "pin_to_retired"] = PydanticField(
        default="follow_new",
        description=(
            "How edge steps that match this vertex on its primary identity behave "
            "afterwards. ``follow_new`` (default) leaves them on the primary, so they "
            "match the new identity. ``pin_to_retired`` rewrites them to select the "
            "demoted secondary identity, preserving the previous matching behaviour "
            "for sources that only carry the old key. Requires ``retire: demote``."
        ),
    )

    @model_validator(mode="after")
    def _validate_endpoint_policy(self) -> IdentityReplacement:
        if self.endpoints == "pin_to_retired" and self.retire != "demote":
            raise ValueError(
                "endpoints: pin_to_retired requires retire: demote — there is no "
                "retired secondary identity to pin to otherwise"
            )
        if self.retire_as is not None and self.retire != "demote":
            raise ValueError("retire_as is only meaningful with retire: demote")
        return self

Attributes

endpoints = PydanticField(default='follow_new', description='How edge steps that match this vertex on its primary identity behave afterwards. ``follow_new`` (default) leaves them on the primary, so they match the new identity. ``pin_to_retired`` rewrites them to select the demoted secondary identity, preserving the previous matching behaviour for sources that only carry the old key. Requires ``retire: demote``.') class-attribute instance-attribute

retire = PydanticField(default='demote', description='What happens to the old identity field-set. ``demote`` turns it into a secondary identity (lookup index follows automatically), ``keep`` leaves the fields as plain properties, ``drop`` removes them. Demotion is downgraded to ``keep`` when the old identity was synthetic (hash / assigned / blank) or already equals the new one.') class-attribute instance-attribute

retire_as = PydanticField(default=None, description='Name for the demoted secondary identity. Defaults to ``retired_identity``. Only meaningful with ``retire: demote``.') class-attribute instance-attribute

to = PydanticField(..., description='The identity policy this vertex should have after the op.') class-attribute instance-attribute

LocalKeySource

Bases: ConfigBaseModel

Where one resource's side-local key comes from, and its namespace tag.

The tag is what keeps records of different sources apart once they fail to fuse: f2 from one source and f2 from another are different entities, and a:f2 / b:f2 say so. It is required so that opting out is a statement, not an omission: tag=None (stored as "", the neutral element, so it survives serialization) keeps the raw value as the local key with no separator — the author's claim that the values are already unique across every source of the class (UUIDs, IRIs, ids the source itself prefixes).

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

class LocalKeySource(ConfigBaseModel):
    """Where one resource's side-local key comes from, and its namespace tag.

    The tag is what keeps records of different sources apart once they fail to
    fuse: ``f2`` from one source and ``f2`` from another are different
    entities, and ``a:f2`` / ``b:f2`` say so. It is required so that opting
    out is a statement, not an omission: ``tag=None`` (stored as ``""``, the
    neutral element, so it survives serialization) keeps the raw value as the
    local key with no separator — the author's claim that the values are
    already unique across every source of the class (UUIDs, IRIs, ids the
    source itself prefixes).
    """

    field: str = PydanticField(
        ...,
        description="RAW doc field carrying the side-local key.",
    )
    tag: str = PydanticField(
        ...,
        description=(
            "Namespace tag: tag 'a' turns 'f2' into 'a:f2'. ``None`` or ``\"\"`` "
            "keeps the raw value, no separator — only for values already "
            "unique across every source of the class."
        ),
    )

    @field_validator("tag", mode="before")
    @classmethod
    def _none_is_the_empty_tag(cls, value: Any) -> Any:
        return "" if value is None else value

    gate: str | None = PydanticField(
        default=None,
        description=(
            "Optional RAW doc field deciding whether this source applies — the "
            "router's discriminator when one resource contributes several "
            "local keys. Omit when ``field`` is empty for the other branches, "
            "which already selects."
        ),
    )
    gate_prefix: str = PydanticField(
        default="",
        description=(
            'Required prefix of the ``gate`` value; ``""`` always passes. '
            "Meaningless without ``gate``."
        ),
    )

    @model_validator(mode="after")
    def _validate_gate(self) -> LocalKeySource:
        if self.gate is None and self.gate_prefix:
            raise ValueError(
                "LocalKeySource: gate_prefix is meaningless without a gate field"
            )
        return self

Attributes

field = PydanticField(..., description='RAW doc field carrying the side-local key.') class-attribute instance-attribute

gate = PydanticField(default=None, description="Optional RAW doc field deciding whether this source applies — the router's discriminator when one resource contributes several local keys. Omit when ``field`` is empty for the other branches, which already selects.") class-attribute instance-attribute

gate_prefix = PydanticField(default='', description='Required prefix of the ``gate`` value; ``""`` always passes. Meaningless without ``gate``.') class-attribute instance-attribute

tag = PydanticField(..., description='Namespace tag: tag \'a\' turns \'f2\' into \'a:f2\'. ``None`` or ``""`` keeps the raw value, no separator — only for values already unique across every source of the class.') class-attribute instance-attribute

LocalKeySpec

Bases: ConfigBaseModel

The canonical fallback identity attribute for non-aligned records.

sources takes the same three shapes as :attr:AlignmentAttribute.sources: one source, a list (one per member, each reading its own column), or a dict keyed by member class (the member decides; the gate is derived from how the resource produces it, so a member-keyed source must not set gate).

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

class LocalKeySpec(ConfigBaseModel):
    """The canonical fallback identity attribute for non-aligned records.

    ``sources`` takes the same three shapes as
    :attr:`AlignmentAttribute.sources`: one source, a list (one per member,
    each reading its own column), or a dict keyed by member class (the member
    decides; the gate is derived from how the resource produces it, so a
    member-keyed source must not set ``gate``).
    """

    name: str = PydanticField(
        default="local_key",
        validation_alias=AliasChoices("name", "into"),
        description=(
            "Canonical fallback property name on the class. ``into`` is "
            "accepted as a legacy alias."
        ),
    )
    sep: str = PydanticField(
        default=":",
        description="Separator between tag and key.",
    )
    sources: dict[
        str, LocalKeySource | list[LocalKeySource] | dict[str, LocalKeySource]
    ] = PydanticField(
        ...,
        min_length=1,
        description=(
            "Per-resource local-key wiring: ``{resource: source}``; ``{resource: "
            "[source, ...]}`` when the members carry different key columns; "
            "``{resource: {member_class: source}}`` when the member decides."
        ),
    )

    @model_validator(mode="after")
    def _validate_member_sources(self) -> LocalKeySpec:
        for resource, entry in self.sources.items():
            if not isinstance(entry, dict):
                continue
            gated = sorted(m for m, src in entry.items() if src.gate is not None)
            if gated:
                raise ValueError(
                    f"LocalKeySpec: member-keyed sources for resource {resource!r} "
                    f"set a gate on {gated}; the member already decides, and the "
                    "gate is derived from how the resource produces it"
                )
        return self

    def sources_for(self, resource: str) -> list[LocalKeySource]:
        """Local-key sources *resource* contributes, in order."""
        source = self.sources.get(resource)
        if source is None:
            return []
        if isinstance(source, LocalKeySource):
            return [source]
        return list(source.values()) if isinstance(source, dict) else list(source)

    def members_for(self, resource: str) -> list[str] | None:
        """Member classes keying *resource*'s sources, or ``None`` if unkeyed."""
        source = self.sources.get(resource)
        return list(source) if isinstance(source, dict) else None

Attributes

name = PydanticField(default='local_key', validation_alias=AliasChoices('name', 'into'), description='Canonical fallback property name on the class. ``into`` is accepted as a legacy alias.') class-attribute instance-attribute

sep = PydanticField(default=':', description='Separator between tag and key.') class-attribute instance-attribute

sources = PydanticField(..., min_length=1, description='Per-resource local-key wiring: ``{resource: source}``; ``{resource: [source, ...]}`` when the members carry different key columns; ``{resource: {member_class: source}}`` when the member decides.') class-attribute instance-attribute

Methods:

members_for(resource)

Member classes keying resource's sources, or None if unkeyed.

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

def members_for(self, resource: str) -> list[str] | None:
    """Member classes keying *resource*'s sources, or ``None`` if unkeyed."""
    source = self.sources.get(resource)
    return list(source) if isinstance(source, dict) else None

sources_for(resource)

Local-key sources resource contributes, in order.

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

def sources_for(self, resource: str) -> list[LocalKeySource]:
    """Local-key sources *resource* contributes, in order."""
    source = self.sources.get(resource)
    if source is None:
        return []
    if isinstance(source, LocalKeySource):
        return [source]
    return list(source.values()) if isinstance(source, dict) else list(source)

MergeEdgesOp

Bases: ConfigBaseModel

Merge source relation names into a canonical relation name.

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

class MergeEdgesOp(ConfigBaseModel):
    """Merge source relation names into a canonical relation name."""

    op: Literal["merge_edges"] = "merge_edges"
    sources: list[str] = PydanticField(
        ...,
        description="Relation names to merge away. Must not include ``into``.",
        min_length=1,
    )
    into: str = PydanticField(
        ...,
        description="Canonical relation name that receives all source relations.",
    )

    @model_validator(mode="after")
    def _validate_sources(self) -> MergeEdgesOp:
        validate_merge_sources(self.sources, self.into, kind="merge_edges")
        return self

Attributes

into = PydanticField(..., description='Canonical relation name that receives all source relations.') class-attribute instance-attribute

op = 'merge_edges' class-attribute instance-attribute

sources = PydanticField(..., description='Relation names to merge away. Must not include ``into``.', min_length=1) class-attribute instance-attribute

MergeManifestsOp

Bases: ConfigBaseModel

Merge two full GraphManifests using explicit equivalence maps.

Binary only — apply via :func:~graflo.architecture.evolution.merge.merge_manifests. Unary :func:~graflo.architecture.evolution.apply.apply_evolution rejects this op.

Empty vertex_equivalences / relation_equivalences yields a disjoint union (schema + resources + bindings), subject to name_conflict / resource_renames.

identity_alignments are applied to the merged union before return (canonical attributes → resource derivations → priority funnel → secondaries). Each entry's vertex must be a declared cluster's merged name.

Equivalences name members in the manifests' own vocabulary (a member may also be spelled by its canonical name when canonical_maps establishes it); a cluster's merged name is into, else the canonical name its members map to, else the one spelling they share.

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

class MergeManifestsOp(ConfigBaseModel):
    """Merge two full ``GraphManifest``s using explicit equivalence maps.

    Binary only — apply via :func:`~graflo.architecture.evolution.merge.merge_manifests`.
    Unary :func:`~graflo.architecture.evolution.apply.apply_evolution` rejects this op.

    Empty ``vertex_equivalences`` / ``relation_equivalences`` yields a disjoint
    union (schema + resources + bindings), subject to ``name_conflict`` /
    ``resource_renames``.

    ``identity_alignments`` are applied to the merged union before return
    (canonical attributes → resource derivations → priority funnel → secondaries).
    Each entry's ``vertex`` must be a declared cluster's merged name.

    Equivalences name members in the manifests' own vocabulary (a member may
    also be spelled by its canonical name when ``canonical_maps`` establishes
    it); a cluster's merged name is ``into``, else the canonical name its
    members map to, else the one spelling they share.
    """

    op: Literal["merge_manifests"] = "merge_manifests"
    vertex_equivalences: list[VertexEquivalence] = PydanticField(
        default_factory=list,
        validation_alias=AliasChoices("vertex_equivalences", "vertices"),
        description=(
            "Explicit vertex equivalences across the two input manifests. "
            "``vertices`` is accepted as a legacy alias."
        ),
    )
    relation_equivalences: list[RelationEquivalence] = PydanticField(
        default_factory=list,
        validation_alias=AliasChoices("relation_equivalences", "relations"),
        description=(
            "Optional relation equivalences across the two input manifests. "
            "``relations`` is accepted as a legacy alias."
        ),
    )
    resource_renames: dict[str, str] = PydanticField(
        default_factory=dict,
        description="Rename map applied to *right* resource names before union.",
    )
    name: str | None = PydanticField(
        default=None,
        description=(
            "Label for the merged manifest and its schema. Unset, the two "
            "sides' names are folded into ``left+right``."
        ),
    )
    target_namespace: str | None = PydanticField(
        default=None,
        description=(
            "Database / graph / space the merged schema deploys into. "
            "Supersedes both sides' ``db_profile.target_namespace`` (so it "
            "also resolves a disagreement between them) and is validated "
            "against the merged ``db_flavor``. Unset, the namespace is "
            "derived from the schema name when deployed."
        ),
    )
    name_conflict: Literal["error", "prefix_right", "union_right"] = PydanticField(
        default="error",
        description=(
            "How to handle name collisions no equivalence covers, on the "
            "right side (vertices, relations, resources, connectors). Vertex "
            "and relation names collide both exactly and when they key alike "
            "under ``canonical_key`` -- ``OrderLine`` and ``order_line`` are "
            "one concept spelled two ways, and merging them into two "
            "unrelated types splits the data silently. ``error`` refuses and "
            "names the equivalences to declare; ``prefix_right`` keeps them "
            "apart under ``r_`` names; ``union_right`` unions by name -- every "
            "exact or near collision becomes a synthesized 1-1 equivalence "
            "into the left spelling, so identity and property reconciliation "
            "apply exactly as to a declared one. ``union_right`` applies to "
            "vertices and relations only (resources and connectors are "
            "addresses, not concepts, so it behaves as ``error`` for them). "
            "``fuse_right`` is accepted as a legacy spelling of "
            "``union_right``; `fuse` is otherwise reserved for records "
            "becoming one node, not for type names."
        ),
    )

    @field_validator("name_conflict", mode="before")
    @classmethod
    def _union_right_was_called_fuse_right(cls, value: Any) -> Any:
        """Accept the pre-rename spelling of ``union_right``.

        The policy unions two type *names*; ``fuse`` everywhere else in the
        contract means two *records* becoming one node
        (``allow_observation_fusion``, identity alignment), so the value was
        renamed. Merge is excluded from the revision vocabulary, so no
        stored change set carries the old spelling -- only authored documents,
        which keep working.
        """
        return "union_right" if value == "fuse_right" else value

    allow_merges: bool = PydanticField(
        default=False,
        description=(
            "Accept a vertex or relation equivalence that collapses more than "
            "one class/relation on a side. A merge is a stated intent — it "
            "fuses entities and can create self-relations — so it must be "
            "acknowledged here rather than inferred from the equivalence list."
        ),
    )
    allow_self_relations: bool = PydanticField(
        default=False,
        description=(
            "Accept a merge whose sources are connected by an edge that "
            "becomes a self-relation once both endpoints land on the same "
            "merged vertex. Forwarded to the per-side ``MergeVerticesOp``."
        ),
    )
    allow_observation_fusion: bool = PydanticField(
        default=False,
        validation_alias=AliasChoices("allow_observation_fusion", "allow_row_fusion"),
        description=(
            "Accept a merge whose sources are produced in one accumulator slot "
            "(the same pipeline level and the same ``role``, or both bare). "
            "Forwarded to the per-side ``CanonicalizeOp``. ``allow_row_fusion`` "
            "is accepted as a legacy alias."
        ),
    )
    allow_dangling_entries: bool = PydanticField(
        default=False,
        description=(
            "Accept canonical map entries that name nothing on the side they "
            "are scoped to, dropping and logging each one instead of refusing "
            "with the list. Set it on a map itself to say the map is broader "
            "than this merge; set it here to say so for both maps at once."
        ),
    )
    identity_alignments: list[IdentityAlignment] = PydanticField(
        default_factory=list,
        description=(
            "Optional identity alignments applied after the schema/resource "
            "union, one per merged class."
        ),
    )
    canonical_maps: dict[Literal["left", "right", "both"], CanonicalMap] = (
        PydanticField(
            default_factory=dict,
            description=(
                "Canonical vocabulary per side. ``left`` / ``right`` apply to "
                "that manifest's own names; ``both`` applies to either side and "
                "to merged names. Merge applies each side's map together "
                "with its equivalences in one step, and refuses when the two "
                "disagree on where a name goes."
            ),
        )
    )

    @model_validator(mode="after")
    def _require_allow_merges_for_nary(self) -> MergeManifestsOp:
        if self.allow_merges:
            return self
        offending: set[str] = set()
        for veq in self.vertex_equivalences:
            if len(veq.left_members) > 1 or len(veq.right_members) > 1:
                offending.add(f"vertex equivalence {_describe_cluster(veq)}")
        for req in self.relation_equivalences:
            if len(req.left_members) > 1 or len(req.right_members) > 1:
                offending.add(f"relation equivalence {_describe_cluster(req)}")
        if offending:
            raise ValueError(
                "merge_manifests: "
                + "; ".join(sorted(offending))
                + " collapses more than one class/relation on a side; a merge "
                "is a stated intent — set allow_merges=True"
            )
        return self

Attributes

allow_dangling_entries = PydanticField(default=False, description='Accept canonical map entries that name nothing on the side they are scoped to, dropping and logging each one instead of refusing with the list. Set it on a map itself to say the map is broader than this merge; set it here to say so for both maps at once.') class-attribute instance-attribute

allow_merges = PydanticField(default=False, description='Accept a vertex or relation equivalence that collapses more than one class/relation on a side. A merge is a stated intent — it fuses entities and can create self-relations — so it must be acknowledged here rather than inferred from the equivalence list.') class-attribute instance-attribute

allow_observation_fusion = PydanticField(default=False, validation_alias=AliasChoices('allow_observation_fusion', 'allow_row_fusion'), description='Accept a merge whose sources are produced in one accumulator slot (the same pipeline level and the same ``role``, or both bare). Forwarded to the per-side ``CanonicalizeOp``. ``allow_row_fusion`` is accepted as a legacy alias.') class-attribute instance-attribute

allow_self_relations = PydanticField(default=False, description='Accept a merge whose sources are connected by an edge that becomes a self-relation once both endpoints land on the same merged vertex. Forwarded to the per-side ``MergeVerticesOp``.') class-attribute instance-attribute

canonical_maps = PydanticField(default_factory=dict, description="Canonical vocabulary per side. ``left`` / ``right`` apply to that manifest's own names; ``both`` applies to either side and to merged names. Merge applies each side's map together with its equivalences in one step, and refuses when the two disagree on where a name goes.") class-attribute instance-attribute

identity_alignments = PydanticField(default_factory=list, description='Optional identity alignments applied after the schema/resource union, one per merged class.') class-attribute instance-attribute

name = PydanticField(default=None, description="Label for the merged manifest and its schema. Unset, the two sides' names are folded into ``left+right``.") class-attribute instance-attribute

name_conflict = PydanticField(default='error', description='How to handle name collisions no equivalence covers, on the right side (vertices, relations, resources, connectors). Vertex and relation names collide both exactly and when they key alike under ``canonical_key`` -- ``OrderLine`` and ``order_line`` are one concept spelled two ways, and merging them into two unrelated types splits the data silently. ``error`` refuses and names the equivalences to declare; ``prefix_right`` keeps them apart under ``r_`` names; ``union_right`` unions by name -- every exact or near collision becomes a synthesized 1-1 equivalence into the left spelling, so identity and property reconciliation apply exactly as to a declared one. ``union_right`` applies to vertices and relations only (resources and connectors are addresses, not concepts, so it behaves as ``error`` for them). ``fuse_right`` is accepted as a legacy spelling of ``union_right``; `fuse` is otherwise reserved for records becoming one node, not for type names.') class-attribute instance-attribute

op = 'merge_manifests' class-attribute instance-attribute

relation_equivalences = PydanticField(default_factory=list, validation_alias=AliasChoices('relation_equivalences', 'relations'), description='Optional relation equivalences across the two input manifests. ``relations`` is accepted as a legacy alias.') class-attribute instance-attribute

resource_renames = PydanticField(default_factory=dict, description='Rename map applied to *right* resource names before union.') class-attribute instance-attribute

target_namespace = PydanticField(default=None, description="Database / graph / space the merged schema deploys into. Supersedes both sides' ``db_profile.target_namespace`` (so it also resolves a disagreement between them) and is validated against the merged ``db_flavor``. Unset, the namespace is derived from the schema name when deployed.") class-attribute instance-attribute

vertex_equivalences = PydanticField(default_factory=list, validation_alias=AliasChoices('vertex_equivalences', 'vertices'), description='Explicit vertex equivalences across the two input manifests. ``vertices`` is accepted as a legacy alias.') class-attribute instance-attribute

MergeVerticesOp

Bases: ConfigBaseModel

Merge source vertices into a single logical name (schema, edges, ingestion).

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

class MergeVerticesOp(ConfigBaseModel):
    """Merge source vertices into a single logical name (schema, edges, ingestion)."""

    op: Literal["merge_vertices"] = "merge_vertices"
    sources: list[str] = PydanticField(
        ...,
        description=(
            "Vertex type names to merge away. Must not include ``into``. "
            "Each name must exist in the schema before the merge."
        ),
        min_length=1,
    )
    into: str = PydanticField(
        ...,
        description=(
            "Resulting vertex type name. If it already exists, source vertices are "
            "merged into it. If it does not exist, a new vertex is built from all sources."
        ),
    )
    allow_self_relations: bool = PydanticField(
        default=False,
        description=(
            "Accept edges whose endpoints both land on ``into``. A self-relation makes "
            "both endpoints share one accumulator slot, so assembly merges observations "
            "that were previously separate nodes. Rejected unless set."
        ),
    )
    allow_observation_fusion: bool = PydanticField(
        default=False,
        validation_alias=AliasChoices("allow_observation_fusion", "allow_row_fusion"),
        description=(
            "Accept resource pipelines whose sources land in one accumulator slot — "
            "the same level and the same ``role``, or both bare. Those steps then "
            "write to one slot, fusing into one node what a single source document "
            "emitted as two vertex observations. Same-level steps with distinct "
            "``role``s never share a slot and need no acknowledgement. Rejected "
            "unless set. ``allow_row_fusion`` is accepted as a legacy alias."
        ),
    )

    @model_validator(mode="after")
    def _validate_sources(self) -> MergeVerticesOp:
        validate_merge_sources(self.sources, self.into, kind="merge_vertices")
        return self

Attributes

allow_observation_fusion = PydanticField(default=False, validation_alias=AliasChoices('allow_observation_fusion', 'allow_row_fusion'), description='Accept resource pipelines whose sources land in one accumulator slot — the same level and the same ``role``, or both bare. Those steps then write to one slot, fusing into one node what a single source document emitted as two vertex observations. Same-level steps with distinct ``role``s never share a slot and need no acknowledgement. Rejected unless set. ``allow_row_fusion`` is accepted as a legacy alias.') class-attribute instance-attribute

allow_self_relations = PydanticField(default=False, description='Accept edges whose endpoints both land on ``into``. A self-relation makes both endpoints share one accumulator slot, so assembly merges observations that were previously separate nodes. Rejected unless set.') class-attribute instance-attribute

into = PydanticField(..., description='Resulting vertex type name. If it already exists, source vertices are merged into it. If it does not exist, a new vertex is built from all sources.') class-attribute instance-attribute

op = 'merge_vertices' class-attribute instance-attribute

sources = PydanticField(..., description='Vertex type names to merge away. Must not include ``into``. Each name must exist in the schema before the merge.', min_length=1) class-attribute instance-attribute

NaturalIdentityTarget

Bases: ConfigBaseModel

Target a natural key: the named properties identify the vertex directly.

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

class NaturalIdentityTarget(ConfigBaseModel):
    """Target a natural key: the named properties identify the vertex directly."""

    mode: Literal["natural"] = "natural"
    identity: list[str] = PydanticField(
        ...,
        description="Property names forming the new primary identity.",
        min_length=1,
    )

Attributes

identity = PydanticField(..., description='Property names forming the new primary identity.', min_length=1) class-attribute instance-attribute

mode = 'natural' class-attribute instance-attribute

ProjectManifestOp

Bases: ConfigBaseModel

Project a manifest to a vertex/edge subgraph with consistent cascade.

Keeps only the requested logical vertices and edges (and optionally resources). All schema, db_profile, ingestion, and bindings references to removed entities are pruned. Inverse edges are not kept by default; list them in keep_edges, or set keep_inverse_edges to keep the declared mirror of every edge that is kept.

With connectivity="induced_prune" (v1 default), when keep_vertices is set, vertex types from that list with no incident surviving edge are dropped.

depth turns keep_vertices from a literal list into seeds for a neighbourhood walk. One rule covers every combination: let E be keep_edges when given and every declared edge otherwise; the survivors are the vertex types within depth hops of a seed along E under direction, and then E restricted to surviving endpoints. So the result is the induced subgraph on the hop ball — an edge between two neighbours survives even though no walk needed it — and keep_edges bounds the walk rather than being overridden by it. Pruning is unchanged: a seed left with no surviving edge is still dropped, whatever the depth.

Edge.by (the third vertex type on an EdgeType.INDIRECT edge) is not part of schema adjacency, so a walk never pulls it in — the same blind spot the flat selection already has.

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

class ProjectManifestOp(ConfigBaseModel):
    """Project a manifest to a vertex/edge subgraph with consistent cascade.

    Keeps only the requested logical vertices and edges (and optionally resources).
    All schema, ``db_profile``, ingestion, and bindings references to removed
    entities are pruned. Inverse edges are **not** kept by default; list them in
    ``keep_edges``, or set ``keep_inverse_edges`` to keep the declared mirror of
    every edge that is kept.

    With ``connectivity=\"induced_prune\"`` (v1 default), when ``keep_vertices`` is
    set, vertex types from that list with no incident surviving edge are dropped.

    ``depth`` turns ``keep_vertices`` from a literal list into seeds for a
    neighbourhood walk. One rule covers every combination: let ``E`` be
    ``keep_edges`` when given and every declared edge otherwise; the survivors are
    the vertex types within ``depth`` hops of a seed along ``E`` under
    ``direction``, and then ``E`` restricted to surviving endpoints. So the result
    is the *induced* subgraph on the hop ball — an edge between two neighbours
    survives even though no walk needed it — and ``keep_edges`` bounds the walk
    rather than being overridden by it. Pruning is unchanged: a seed left with no
    surviving edge is still dropped, whatever the depth.

    ``Edge.by`` (the third vertex type on an ``EdgeType.INDIRECT`` edge) is not part
    of schema adjacency, so a walk never pulls it in — the same blind spot the flat
    selection already has.
    """

    op: Literal["project_manifest"] = "project_manifest"
    keep_vertices: list[str] | None = PydanticField(
        default=None,
        description="Vertex type names to retain (after induced connectivity pruning).",
    )
    keep_edges: list[EdgeSelector] | None = PydanticField(
        default=None,
        description="Edge triples ``(source, target, relation)`` to retain.",
    )
    connectivity: Literal["induced_prune"] = PydanticField(
        default="induced_prune",
        description="How to interpret ``keep_vertices`` relative to surviving edges.",
    )
    depth: int = PydanticField(
        default=0,
        ge=0,
        description=(
            "Hops to expand ``keep_vertices`` by before the induced slice. "
            "``0`` (default) keeps the literal list."
        ),
    )
    direction: EdgeDirection = PydanticField(
        default=EdgeDirection.ANY,
        description=(
            "Orientation followed when expanding by ``depth``. Edges declared "
            "``directed: false`` are followed both ways regardless. Ignored when "
            "``depth`` is 0."
        ),
    )
    keep_resources: list[str] | None = PydanticField(
        default=None,
        description="Optional ingestion resource names to retain after graph slice.",
    )
    keep_inverse_edges: bool = PydanticField(
        default=False,
        description=(
            "With ``keep_edges``: also keep the declared mirror ``(T, S, inv)`` of "
            "each kept ``(S, T, r)``, so a materialized pair survives as a pair. "
            "Without ``keep_edges`` every edge between surviving vertices is kept "
            "already."
        ),
    )
    strict: bool = PydanticField(
        default=True,
        description="When True, unknown vertex/edge selectors raise ``ValueError``.",
    )

    @model_validator(mode="after")
    def _validate_projection_selectors(self) -> ProjectManifestOp:
        if not self.keep_vertices and not self.keep_edges:
            raise ValueError(
                "project_manifest requires at least one of keep_vertices or keep_edges"
            )
        if self.keep_vertices and len(self.keep_vertices) != len(
            set(self.keep_vertices)
        ):
            raise ValueError("keep_vertices entries must be unique")
        if self.keep_edges:
            edge_ids = [selector.edge_id() for selector in self.keep_edges]
            if len(edge_ids) != len(set(edge_ids)):
                raise ValueError(
                    "keep_edges entries must be unique by (source, target, relation)"
                )
        if self.depth > 0 and not self.keep_vertices:
            # Without seeds `keep_vertices=None` already means "every vertex type",
            # so there is nothing for a walk to expand — the request is a mistake
            # rather than a no-op, and saying so beats silently ignoring `depth`.
            raise ValueError("project_manifest: depth > 0 requires keep_vertices")
        return self

Attributes

connectivity = PydanticField(default='induced_prune', description='How to interpret ``keep_vertices`` relative to surviving edges.') class-attribute instance-attribute

depth = PydanticField(default=0, ge=0, description='Hops to expand ``keep_vertices`` by before the induced slice. ``0`` (default) keeps the literal list.') class-attribute instance-attribute

direction = PydanticField(default=EdgeDirection.ANY, description='Orientation followed when expanding by ``depth``. Edges declared ``directed: false`` are followed both ways regardless. Ignored when ``depth`` is 0.') class-attribute instance-attribute

keep_edges = PydanticField(default=None, description='Edge triples ``(source, target, relation)`` to retain.') class-attribute instance-attribute

keep_inverse_edges = PydanticField(default=False, description='With ``keep_edges``: also keep the declared mirror ``(T, S, inv)`` of each kept ``(S, T, r)``, so a materialized pair survives as a pair. Without ``keep_edges`` every edge between surviving vertices is kept already.') class-attribute instance-attribute

keep_resources = PydanticField(default=None, description='Optional ingestion resource names to retain after graph slice.') class-attribute instance-attribute

keep_vertices = PydanticField(default=None, description='Vertex type names to retain (after induced connectivity pruning).') class-attribute instance-attribute

op = 'project_manifest' class-attribute instance-attribute

strict = PydanticField(default=True, description='When True, unknown vertex/edge selectors raise ``ValueError``.') class-attribute instance-attribute

PropertyEquivalence

Bases: ConfigBaseModel

Align a property from the left and/or right member(s) onto a canonical name.

At least one of left / right must be set. A bare string applies to every member declared on that side of the owning :class:VertexEquivalence; a {member: field} dict maps per member, for when members are not aligned under the same source field name.

Exact-name matches do not need a :class:PropertyEquivalence: after boundary rename, merge_vertex_models unions fields by spelling, so a property present under the same name on every member fuses for free. Declare an equivalence only to rename, to pick a different into, or to flag identity=True.

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

class PropertyEquivalence(ConfigBaseModel):
    """Align a property from the left and/or right member(s) onto a canonical name.

    At least one of ``left`` / ``right`` must be set. A bare string applies to
    every member declared on that side of the owning
    :class:`VertexEquivalence`; a ``{member: field}`` dict maps per member,
    for when members are not aligned under the same source field name.

    Exact-name matches do **not** need a :class:`PropertyEquivalence`: after
    boundary rename, ``merge_vertex_models`` unions fields by spelling, so a
    property present under the same name on every member fuses for free.
    Declare an equivalence only to rename, to pick a different ``into``, or to
    flag ``identity=True``.
    """

    left: str | dict[str, str] | None = PydanticField(
        default=None,
        description=(
            "Field name on the left member(s): a bare string applies to every "
            "left member of the owning equivalence, a ``{member: field}`` dict "
            "maps per member."
        ),
    )
    right: str | dict[str, str] | None = PydanticField(
        default=None,
        description="Same shape as ``left``, for the right member(s).",
    )
    into: str = PydanticField(
        ...,
        description="Canonical property name on the merged vertex.",
    )
    identity: bool = PydanticField(
        default=False,
        description=(
            "When True and ``VertexEquivalence.identity`` is unset, include ``into`` "
            "in the derived identity list after merge."
        ),
    )

    @model_validator(mode="after")
    def _require_side(self) -> PropertyEquivalence:
        if self.left is None and self.right is None:
            raise ValueError(
                "PropertyEquivalence requires at least one of left or right"
            )
        for side, spec in (("left", self.left), ("right", self.right)):
            if isinstance(spec, dict) and not spec:
                raise ValueError(
                    f"PropertyEquivalence: {side} is an empty per-member map"
                )
        return self

Attributes

identity = PydanticField(default=False, description='When True and ``VertexEquivalence.identity`` is unset, include ``into`` in the derived identity list after merge.') class-attribute instance-attribute

into = PydanticField(..., description='Canonical property name on the merged vertex.') class-attribute instance-attribute

left = PydanticField(default=None, description='Field name on the left member(s): a bare string applies to every left member of the owning equivalence, a ``{member: field}`` dict maps per member.') class-attribute instance-attribute

right = PydanticField(default=None, description='Same shape as ``left``, for the right member(s).') class-attribute instance-attribute

RelationEquivalence

Bases: ConfigBaseModel

Collapse one or more left relations and one or more right relations onto one name.

Shares the left / right n-ary shape of :class:VertexEquivalence: a bare name is a 1-1 equivalence, a list is a merge and requires MergeManifestsOp.allow_merges=True.

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

class RelationEquivalence(ConfigBaseModel):
    """Collapse one or more left relations and one or more right relations onto one name.

    Shares the ``left`` / ``right`` n-ary shape of :class:`VertexEquivalence`:
    a bare name is a 1-1 equivalence, a list is a merge and requires
    ``MergeManifestsOp.allow_merges=True``.
    """

    left: str | list[str] = PydanticField(
        ..., description="One or more left relation names."
    )
    right: str | list[str] = PydanticField(
        ..., description="One or more right relation names."
    )
    into: str | None = PydanticField(
        default=None,
        description=(
            "Merged relation name. Omitted, the name comes from the canonical "
            "map that maps a member, or from the one spelling every member shares."
        ),
    )

    @property
    def left_members(self) -> list[str]:
        return _member_list(self.left)

    @property
    def right_members(self) -> list[str]:
        return _member_list(self.right)

    def members(self, side: Literal["left", "right"]) -> list[str]:
        return self.left_members if side == "left" else self.right_members

    @model_validator(mode="after")
    def _validate_members(self) -> RelationEquivalence:
        for side, members in (
            ("left", self.left_members),
            ("right", self.right_members),
        ):
            if not members:
                raise ValueError(
                    f"RelationEquivalence: {side} must name at least one relation"
                )
            if len(members) != len(set(members)):
                raise ValueError(
                    f"RelationEquivalence: {side} lists a relation more than once: {members}"
                )
        return self

Attributes

into = PydanticField(default=None, description='Merged relation name. Omitted, the name comes from the canonical map that maps a member, or from the one spelling every member shares.') class-attribute instance-attribute

left = PydanticField(..., description='One or more left relation names.') class-attribute instance-attribute

left_members property

right = PydanticField(..., description='One or more right relation names.') class-attribute instance-attribute

right_members property

Methods:

members(side)

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

def members(self, side: Literal["left", "right"]) -> list[str]:
    return self.left_members if side == "left" else self.right_members

RemoveEdgeIndexesOp

Bases: ConfigBaseModel

Withdraw authored indexes from edge physical specs, addressed by field list.

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

class RemoveEdgeIndexesOp(ConfigBaseModel):
    """Withdraw authored indexes from edge physical specs, addressed by field list."""

    op: Literal["remove_edge_indexes"] = "remove_edge_indexes"
    edges: list[EdgeIndexEntry] = PydanticField(
        ...,
        description="Per-spec indexes to remove (``fields`` on each entry).",
        min_length=1,
    )

    @model_validator(mode="after")
    def _validate_entries(self) -> RemoveEdgeIndexesOp:
        _validate_edge_index_entries(
            self.edges, kind="remove_edge_indexes", carries="fields"
        )
        return self

Attributes

edges = PydanticField(..., description='Per-spec indexes to remove (``fields`` on each entry).', min_length=1) class-attribute instance-attribute

op = 'remove_edge_indexes' class-attribute instance-attribute

RemoveEdgePropertiesOp

Bases: ConfigBaseModel

Remove edge properties for each relation across schema/profile/ingestion.

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

class RemoveEdgePropertiesOp(ConfigBaseModel):
    """Remove edge properties for each relation across schema/profile/ingestion."""

    op: Literal["remove_edge_properties"] = "remove_edge_properties"
    removals: dict[str, list[str]] = PydanticField(
        ...,
        description=(
            "Per-relation edge field removals: ``{relation_name: [field_name, ...]}``."
        ),
        min_length=1,
    )

Attributes

op = 'remove_edge_properties' class-attribute instance-attribute

removals = PydanticField(..., description='Per-relation edge field removals: ``{relation_name: [field_name, ...]}``.', min_length=1) class-attribute instance-attribute

RemoveEdgesOp

Bases: ConfigBaseModel

Remove logical edges from schema, profile, and ingestion selectors.

Two addressing forms, combinable in one op. relations removes a relation on every endpoint pair it occurs on. edges removes exactly the named triples, which is the only way to remove one pair of several sharing a relation, or an edge with no relation set at all.

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

class RemoveEdgesOp(ConfigBaseModel):
    """Remove logical edges from schema, profile, and ingestion selectors.

    Two addressing forms, combinable in one op. ``relations`` removes a relation
    on every endpoint pair it occurs on. ``edges`` removes exactly the named
    triples, which is the only way to remove one pair of several sharing a
    relation, or an edge with no relation set at all.
    """

    op: Literal["remove_edges"] = "remove_edges"
    relations: list[str] = PydanticField(
        default_factory=list,
        description="Relation names to remove from edge definitions and references.",
    )
    edges: list[EdgeSelector] = PydanticField(
        default_factory=list,
        description="Edge triples ``(source, target, relation)`` to remove.",
    )

    @model_validator(mode="after")
    def _validate_targets(self) -> RemoveEdgesOp:
        if not self.relations and not self.edges:
            raise ValueError("remove_edges requires at least one of relations or edges")
        _validate_unique_edge_selectors(self.edges, kind="remove_edges")
        return self

Attributes

edges = PydanticField(default_factory=list, description='Edge triples ``(source, target, relation)`` to remove.') class-attribute instance-attribute

op = 'remove_edges' class-attribute instance-attribute

relations = PydanticField(default_factory=list, description='Relation names to remove from edge definitions and references.') class-attribute instance-attribute

RemoveResourcesOp

Bases: ConfigBaseModel

Remove ingestion resources and the bindings that wired them.

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

class RemoveResourcesOp(ConfigBaseModel):
    """Remove ingestion resources and the bindings that wired them."""

    op: Literal["remove_resources"] = "remove_resources"
    names: list[str] = PydanticField(
        ...,
        description="Resource names to remove.",
        min_length=1,
    )

Attributes

names = PydanticField(..., description='Resource names to remove.', min_length=1) class-attribute instance-attribute

op = 'remove_resources' class-attribute instance-attribute

RemoveSecondaryIdentitiesOp

Bases: ConfigBaseModel

Withdraw alternate lookup keys, dropping their derived indexes.

Rejected when a surviving edge step still selects the removed field-set — that step would have no way to resolve its endpoint.

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

class RemoveSecondaryIdentitiesOp(ConfigBaseModel):
    """Withdraw alternate lookup keys, dropping their derived indexes.

    Rejected when a surviving edge step still selects the removed field-set — that
    step would have no way to resolve its endpoint.
    """

    op: Literal["remove_secondary_identities"] = "remove_secondary_identities"
    removals: dict[str, list[str | list[str]]] = PydanticField(
        ...,
        description=(
            "Per-vertex secondary identities to withdraw, addressed by name or by "
            "field list: ``{vertex_name: [name | [field, ...], ...]}``."
        ),
        min_length=1,
    )

Attributes

op = 'remove_secondary_identities' class-attribute instance-attribute

removals = PydanticField(..., description='Per-vertex secondary identities to withdraw, addressed by name or by field list: ``{vertex_name: [name | [field, ...], ...]}``.', min_length=1) class-attribute instance-attribute

RemoveVertexIndexesOp

Bases: ConfigBaseModel

Withdraw authored vertex indexes, addressed by field list.

Indexes derived from secondary_identities are not removable here — they would be re-registered by the next finish_init. Use :class:RemoveSecondaryIdentitiesOp for those.

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

class RemoveVertexIndexesOp(ConfigBaseModel):
    """Withdraw authored vertex indexes, addressed by field list.

    Indexes derived from ``secondary_identities`` are not removable here — they would
    be re-registered by the next ``finish_init``. Use
    :class:`RemoveSecondaryIdentitiesOp` for those.
    """

    op: Literal["remove_vertex_indexes"] = "remove_vertex_indexes"
    indexes: dict[
        str,
        Annotated[
            list[Annotated[list[str], PydanticField(min_length=1)]],
            PydanticField(min_length=1),
        ],
    ] = PydanticField(
        ...,
        description="``{vertex_name: [[field, ...], ...]}``.",
        min_length=1,
    )

Attributes

indexes = PydanticField(..., description='``{vertex_name: [[field, ...], ...]}``.', min_length=1) class-attribute instance-attribute

op = 'remove_vertex_indexes' class-attribute instance-attribute

RemoveVertexPropertiesOp

Bases: ConfigBaseModel

Remove vertex properties and propagate pruning to ingestion/db profile references.

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

class RemoveVertexPropertiesOp(ConfigBaseModel):
    """Remove vertex properties and propagate pruning to ingestion/db profile references."""

    op: Literal["remove_vertex_properties"] = "remove_vertex_properties"
    removals: dict[str, list[str]] = PydanticField(
        ...,
        description=(
            "Per-vertex field removal map: ``{vertex_name: [field_name, ...]}``."
        ),
        min_length=1,
    )

Attributes

op = 'remove_vertex_properties' class-attribute instance-attribute

removals = PydanticField(..., description='Per-vertex field removal map: ``{vertex_name: [field_name, ...]}``.', min_length=1) class-attribute instance-attribute

RemoveVerticesOp

Bases: ConfigBaseModel

Remove logical vertices and cascade: edges, ingestion resources, bindings.

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

class RemoveVerticesOp(ConfigBaseModel):
    """Remove logical vertices and cascade: edges, ingestion resources, bindings."""

    op: Literal["remove_vertices"] = "remove_vertices"
    names: list[str] = PydanticField(
        ...,
        description="Vertex type names to remove from the schema.",
        min_length=1,
    )

Attributes

names = PydanticField(..., description='Vertex type names to remove from the schema.', min_length=1) class-attribute instance-attribute

op = 'remove_vertices' class-attribute instance-attribute

RenameEdgePropertiesOp

Bases: ConfigBaseModel

Rename edge properties for each relation across schema/profile/ingestion.

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

class RenameEdgePropertiesOp(ConfigBaseModel):
    """Rename edge properties for each relation across schema/profile/ingestion."""

    op: Literal["rename_edge_properties"] = "rename_edge_properties"
    renames: dict[str, dict[str, str]] = PydanticField(
        ...,
        description=(
            "Per-relation edge field rename map: "
            "``{relation_name: {old_field: new_field}}``."
        ),
        min_length=1,
    )

Attributes

op = 'rename_edge_properties' class-attribute instance-attribute

renames = PydanticField(..., description='Per-relation edge field rename map: ``{relation_name: {old_field: new_field}}``.', min_length=1) class-attribute instance-attribute

RenameRelationsOp

Bases: ConfigBaseModel

Rename logical edge relation names across schema and ingestion.

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

class RenameRelationsOp(ConfigBaseModel):
    """Rename logical edge relation names across schema and ingestion."""

    op: Literal["rename_relations"] = "rename_relations"
    renames: dict[str, str] = PydanticField(
        ...,
        validation_alias=AliasChoices("renames", "relations"),
        description=(
            "Relation rename map: ``{old_relation: new_relation}``. Must be "
            "injective. ``relations`` is accepted as a legacy alias."
        ),
        min_length=1,
    )

    @model_validator(mode="after")
    def _reject_collapsing_map(self) -> RenameRelationsOp:
        validate_rename_map_is_injective(
            self.renames,
            kind="rename_relations",
            merge_hint="MergeEdgesOp(sources=[...], into=...)",
        )
        return self

Attributes

op = 'rename_relations' class-attribute instance-attribute

renames = PydanticField(..., validation_alias=AliasChoices('renames', 'relations'), description='Relation rename map: ``{old_relation: new_relation}``. Must be injective. ``relations`` is accepted as a legacy alias.', min_length=1) class-attribute instance-attribute

RenameResourcesOp

Bases: ConfigBaseModel

Rename ingestion resource names and bindings references.

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

class RenameResourcesOp(ConfigBaseModel):
    """Rename ingestion resource names and bindings references."""

    op: Literal["rename_resources"] = "rename_resources"
    renames: dict[str, str] = PydanticField(
        ...,
        validation_alias=AliasChoices("renames", "resources"),
        description=(
            "Ingestion resource rename map: ``{old_resource: new_resource}``. Must "
            "be injective. ``resources`` is accepted as a legacy alias."
        ),
        min_length=1,
    )

    @model_validator(mode="after")
    def _reject_collapsing_map(self) -> RenameResourcesOp:
        # IngestionModel already rejects duplicate resource names, so a collapsing
        # map fails downstream anyway — but with a message about the model rather
        # than about the op the author actually wrote.
        validate_rename_map_is_injective(
            self.renames,
            kind="rename_resources",
            merge_hint="MergeManifestsOp with explicit resource_renames",
        )
        return self

Attributes

op = 'rename_resources' class-attribute instance-attribute

renames = PydanticField(..., validation_alias=AliasChoices('renames', 'resources'), description='Ingestion resource rename map: ``{old_resource: new_resource}``. Must be injective. ``resources`` is accepted as a legacy alias.', min_length=1) class-attribute instance-attribute

RenameVertexPropertiesOp

Bases: ConfigBaseModel

Rename vertex properties (and identity references) and propagate to ingestion.

renames maps each vertex name to a per-vertex {old_field: new_field} map. Schema-side: rewrites Field.name, vertex.identity, and any DB profile structures that reference field names (vertex_indexes, edge_specs.indexes). Ingestion-side: rewrites VertexActor.from so the doc still uses the OLD field name (injecting {new_field: old_field} when missing), rewrites TransformActor.rename values that target a renamed vertex field, and updates Resource.extra_weights / edge.vertex_weights (:class:~graflo.architecture.graph_types.Weight fields, map, and filter keys that address vertex observation columns).

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

class RenameVertexPropertiesOp(ConfigBaseModel):
    """Rename vertex properties (and identity references) and propagate to ingestion.

    ``renames`` maps each vertex name to a per-vertex ``{old_field: new_field}`` map.
    Schema-side: rewrites ``Field.name``, ``vertex.identity``, and any DB profile
    structures that reference field names (``vertex_indexes``, ``edge_specs.indexes``).
    Ingestion-side: rewrites ``VertexActor.from`` so the doc still uses the OLD field
    name (injecting ``{new_field: old_field}`` when missing), rewrites
    ``TransformActor.rename`` values that target a renamed vertex field, and updates
    ``Resource.extra_weights`` / ``edge.vertex_weights`` (:class:`~graflo.architecture.graph_types.Weight`
    ``fields``, ``map``, and ``filter`` keys that address vertex observation columns).
    """

    op: Literal["rename_vertex_properties"] = "rename_vertex_properties"
    renames: dict[str, dict[str, str]] = PydanticField(
        ...,
        description=(
            "Per-vertex field rename map: ``{vertex_name: {old_field: new_field}}``."
        ),
        min_length=1,
    )

Attributes

op = 'rename_vertex_properties' class-attribute instance-attribute

renames = PydanticField(..., description='Per-vertex field rename map: ``{vertex_name: {old_field: new_field}}``.', min_length=1) class-attribute instance-attribute

RenameVerticesOp

Bases: ConfigBaseModel

Rename logical vertex names across schema, ingestion, and bindings.

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

class RenameVerticesOp(ConfigBaseModel):
    """Rename logical vertex names across schema, ingestion, and bindings."""

    op: Literal["rename_vertices"] = "rename_vertices"
    renames: dict[str, str] = PydanticField(
        ...,
        validation_alias=AliasChoices("renames", "vertices"),
        description=(
            "Vertex rename map: ``{old_vertex: new_vertex}``. Must be injective. "
            "``vertices`` is accepted as a legacy alias."
        ),
        min_length=1,
    )

    @model_validator(mode="after")
    def _reject_collapsing_map(self) -> RenameVerticesOp:
        validate_rename_map_is_injective(
            self.renames,
            kind="rename_vertices",
            merge_hint="MergeVerticesOp(sources=[...], into=...)",
        )
        return self

Attributes

op = 'rename_vertices' class-attribute instance-attribute

renames = PydanticField(..., validation_alias=AliasChoices('renames', 'vertices'), description='Vertex rename map: ``{old_vertex: new_vertex}``. Must be injective. ``vertices`` is accepted as a legacy alias.', min_length=1) class-attribute instance-attribute

ReplaceEdgeIdentitiesOp

Bases: ConfigBaseModel

Replace the uniqueness keys of logical edges.

The edge-side counterpart of :class:ReplaceIdentityOp. There is no retire policy: edge identities have no lookup plane to demote into. Non-endpoint tokens are merged into edge properties by Edge.finish_init, as with authored identities.

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

class ReplaceEdgeIdentitiesOp(ConfigBaseModel):
    """Replace the uniqueness keys of logical edges.

    The edge-side counterpart of :class:`ReplaceIdentityOp`. There is no retire policy:
    edge identities have no lookup plane to demote into. Non-endpoint tokens are merged
    into edge ``properties`` by ``Edge.finish_init``, as with authored identities.
    """

    op: Literal["replace_edge_identities"] = "replace_edge_identities"
    edges: list[EdgeIdentitiesEntry] = PydanticField(
        ...,
        description="Per-edge replacement uniqueness keys.",
        min_length=1,
    )

    @model_validator(mode="after")
    def _validate_unique_selectors(self) -> ReplaceEdgeIdentitiesOp:
        edge_ids = [entry.edge_id() for entry in self.edges]
        if len(edge_ids) != len(set(edge_ids)):
            raise ValueError(
                "replace_edge_identities entries must be unique by "
                "(source, target, relation)"
            )
        return self

Attributes

edges = PydanticField(..., description='Per-edge replacement uniqueness keys.', min_length=1) class-attribute instance-attribute

op = 'replace_edge_identities' class-attribute instance-attribute

ReplaceIdentityOp

Bases: ConfigBaseModel

Replace the identity policy of one or more vertices.

Covers both a change of identity fields and a change of identity mode (natural / hash / assigned / blank), because the cascade is the same in either case: the field-set that upserts changes, and everything that referenced the old one must be repointed or retired.

Not covered: blank vertices cannot retire by demotion (they cannot declare secondary identities at all), and a no-op replacement does not bump the version.

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

class ReplaceIdentityOp(ConfigBaseModel):
    """Replace the identity policy of one or more vertices.

    Covers both a change of identity *fields* and a change of identity *mode*
    (``natural`` / ``hash`` / ``assigned`` / ``blank``), because the cascade is the
    same in either case: the field-set that upserts changes, and everything that
    referenced the old one must be repointed or retired.

    Not covered: ``blank`` vertices cannot retire by demotion (they cannot declare
    secondary identities at all), and a no-op replacement does not bump the version.
    """

    op: Literal["replace_identity"] = "replace_identity"
    replacements: dict[str, IdentityReplacement] = PydanticField(
        ...,
        validation_alias=AliasChoices("replacements", "vertices"),
        description=(
            "Per-vertex identity replacement: ``{vertex_name: replacement}``. "
            "``vertices`` is accepted as a legacy alias."
        ),
        min_length=1,
    )

Attributes

op = 'replace_identity' class-attribute instance-attribute

replacements = PydanticField(..., validation_alias=AliasChoices('replacements', 'vertices'), description='Per-vertex identity replacement: ``{vertex_name: replacement}``. ``vertices`` is accepted as a legacy alias.', min_length=1) class-attribute instance-attribute

RetargetEdgesOp

Bases: ConfigBaseModel

Change which vertex types an edge connects, preserving everything else.

Remove-plus-add would lose the edge's properties, uniqueness keys, directed flag, and its db_profile physical spec. Retargeting rewrites the EdgeId everywhere it is keyed instead: edge config, physical specs, and pipeline edge steps.

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

class RetargetEdgesOp(ConfigBaseModel):
    """Change which vertex types an edge connects, preserving everything else.

    Remove-plus-add would lose the edge's properties, uniqueness keys, ``directed``
    flag, and its ``db_profile`` physical spec. Retargeting rewrites the ``EdgeId``
    everywhere it is keyed instead: edge config, physical specs, and pipeline edge steps.
    """

    op: Literal["retarget_edges"] = "retarget_edges"
    edges: list[EdgeRetargetEntry] = PydanticField(
        ...,
        description="Per-edge endpoint retargeting.",
        min_length=1,
    )

    @model_validator(mode="after")
    def _validate_unique_selectors(self) -> RetargetEdgesOp:
        edge_ids = [entry.edge_id() for entry in self.edges]
        if len(edge_ids) != len(set(edge_ids)):
            raise ValueError(
                "retarget_edges entries must be unique by (source, target, relation)"
            )
        retargeted = [entry.retargeted_edge_id() for entry in self.edges]
        if len(retargeted) != len(set(retargeted)):
            raise ValueError(
                "retarget_edges entries must not collide after retargeting"
            )
        return self

Attributes

edges = PydanticField(..., description='Per-edge endpoint retargeting.', min_length=1) class-attribute instance-attribute

op = 'retarget_edges' class-attribute instance-attribute

RetractEdgeInversesOp

Bases: ConfigBaseModel

Withdraw declared inverses, addressed by relation name.

A paired relation retracts its whole pair (either side names it); a symmetric relation retracts its own declaration. Refused while a native inverse still realizes a pair: that would leave the database maintaining a type the schema no longer names. Explicit inverse edges are ordinary edges and survive a retraction; so does directed: false.

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

class RetractEdgeInversesOp(ConfigBaseModel):
    """Withdraw declared inverses, addressed by relation name.

    A paired relation retracts its whole pair (either side names it); a
    symmetric relation retracts its own declaration. Refused while a native
    inverse still realizes a pair: that would leave the database maintaining a
    type the schema no longer names. Explicit inverse edges are ordinary edges
    and survive a retraction; so does ``directed: false``.
    """

    op: Literal["retract_edge_inverses"] = "retract_edge_inverses"
    relations: list[str] = PydanticField(
        ...,
        description="Relations whose declaration is withdrawn (either side of a pair).",
        min_length=1,
    )

Attributes

op = 'retract_edge_inverses' class-attribute instance-attribute

relations = PydanticField(..., description='Relations whose declaration is withdrawn (either side of a pair).', min_length=1) class-attribute instance-attribute

SanitizeOp

Bases: ConfigBaseModel

Apply DB-flavor-specific name/field sanitization to a manifest.

Merges (in order):

1. Storage-name sanitization on DatabaseProfile (vertex storage names + edge relation names) against the flavor's reserved-words set.
2. Vertex field rename for fields whose names are reserved words.
3. For TigerGraph, normalize identity fields across edges that share a relation (TigerGraph requires consistent source/target indexes per relation).

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

class SanitizeOp(ConfigBaseModel):
    """Apply DB-flavor-specific name/field sanitization to a manifest.

    Merges (in order):

    1. Storage-name sanitization on ``DatabaseProfile`` (vertex storage names + edge
       relation names) against the flavor's reserved-words set.
    2. Vertex field rename for fields whose names are reserved words.
    3. For TigerGraph, normalize identity fields across edges that share a relation
       (TigerGraph requires consistent source/target indexes per relation).
    """

    op: Literal["sanitize"] = "sanitize"
    db_flavor: DBType = PydanticField(
        ...,
        description="Target database flavor whose reserved words/constraints drive the sanitization.",
    )
    reserved_words: list[str] | None = PydanticField(
        default=None,
        description=(
            "Optional override for the flavor's reserved words. "
            "When unset, ``graflo.db.util.load_reserved_words(db_flavor)`` is used."
        ),
    )

Attributes

db_flavor = PydanticField(..., description='Target database flavor whose reserved words/constraints drive the sanitization.') class-attribute instance-attribute

op = 'sanitize' class-attribute instance-attribute

reserved_words = PydanticField(default=None, description="Optional override for the flavor's reserved words. When unset, ``graflo.db.util.load_reserved_words(db_flavor)`` is used.") class-attribute instance-attribute

SetBindingsOp

Bases: ConfigBaseModel

Replace the whole bindings block.

The bindings block had no op at all, so diff_manifests could only report it as inexpressible and a change set that touched it was not replayable -- which is why a merge that unions two bindings registries could not be recorded as a commit.

Wholesale rather than granular (add/remove/rename a connector) because that is what the diff needs to say: the block became this. The cost is coarseness in a three-way merge -- the whole block is one slot, so two independent bindings edits conflict where granular ops would merge. Granular connector ops can be added later without changing this one's meaning.

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

class SetBindingsOp(ConfigBaseModel):
    """Replace the whole ``bindings`` block.

    The bindings block had **no op at all**, so ``diff_manifests`` could only
    report it as inexpressible and a change set that touched it was not
    replayable -- which is why a merge that unions two bindings registries
    could not be recorded as a commit.

    Wholesale rather than granular (add/remove/rename a connector) because that
    is what the diff needs to say: the block became this. The cost is
    coarseness in a three-way merge -- the whole block is one slot, so two
    independent bindings edits conflict where granular ops would merge. Granular
    connector ops can be added later without changing this one's meaning.
    """

    op: Literal["set_bindings"] = "set_bindings"
    bindings: Bindings | None = PydanticField(
        default=None,
        description=(
            "The bindings block after the op. ``None`` removes it, which is how "
            "the op stays total: every before/after pair is expressible."
        ),
    )

Attributes

bindings = PydanticField(default=None, description='The bindings block after the op. ``None`` removes it, which is how the op stays total: every before/after pair is expressible.') class-attribute instance-attribute

op = 'set_bindings' class-attribute instance-attribute

SetDbProfileOp

Bases: ConfigBaseModel

Replace the whole db_profile of the schema block.

vertex_indexes and each edge spec's indexes already have four authoring ops; nothing else on the profile had any, so db_flavor, target_namespace, vertex_storage_names, default_property_values and the non-index parts of edge_specs were inexpressible -- and they are part of the content hash, so a change set that moved one of them could not replay.

This op carries the whole profile, indexes included, and therefore subsumes the index ops when it is emitted; the differ emits it instead of them rather than alongside, so the two can never fight over ordering. When only indexes differ, the index ops are still what gets emitted -- they say more about intent and merge at a finer slot.

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

class SetDbProfileOp(ConfigBaseModel):
    """Replace the whole ``db_profile`` of the schema block.

    ``vertex_indexes`` and each edge spec's ``indexes`` already have four
    authoring ops; nothing else on the profile had any, so ``db_flavor``,
    ``target_namespace``, ``vertex_storage_names``, ``default_property_values``
    and the non-index parts of ``edge_specs`` were inexpressible -- and they are
    part of the content hash, so a change set that moved one of them could not
    replay.

    This op carries the **whole** profile, indexes included, and therefore
    subsumes the index ops when it is emitted; the differ emits it *instead of*
    them rather than alongside, so the two can never fight over ordering. When
    only indexes differ, the index ops are still what gets emitted -- they say
    more about intent and merge at a finer slot.
    """

    op: Literal["set_db_profile"] = "set_db_profile"
    profile: DatabaseProfile = PydanticField(
        ...,
        description="The database profile after the op, replacing the current one.",
    )

Attributes

op = 'set_db_profile' class-attribute instance-attribute

profile = PydanticField(..., description='The database profile after the op, replacing the current one.') class-attribute instance-attribute

SetEdgeDirectedOp

Bases: ConfigBaseModel

Set the directed flag on logical edges.

Small, but load-bearing for replay: directed decides what :class:AddInverseEdgesOp is allowed to duplicate, so an un-authorable flag makes inverse-edge change sets non-replayable.

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

class SetEdgeDirectedOp(ConfigBaseModel):
    """Set the ``directed`` flag on logical edges.

    Small, but load-bearing for replay: ``directed`` decides what
    :class:`AddInverseEdgesOp` is allowed to duplicate, so an un-authorable flag makes
    inverse-edge change sets non-replayable.
    """

    op: Literal["set_edge_directed"] = "set_edge_directed"
    edges: list[EdgeSelector] = PydanticField(
        ...,
        description="Edge triples whose ``directed`` flag changes.",
        min_length=1,
    )
    directed: bool = PydanticField(
        ...,
        description="Value applied to every selected edge.",
    )

    @model_validator(mode="after")
    def _validate_unique_selectors(self) -> SetEdgeDirectedOp:
        _validate_unique_edge_selectors(self.edges, kind="set_edge_directed")
        return self

Attributes

directed = PydanticField(..., description='Value applied to every selected edge.') class-attribute instance-attribute

edges = PydanticField(..., description='Edge triples whose ``directed`` flag changes.', min_length=1) class-attribute instance-attribute

op = 'set_edge_directed' class-attribute instance-attribute

SetEdgeSemanticsOp

Bases: ConfigBaseModel

Ground edge relations in an external vocabulary.

The vertex op's counterpart. Relations carry as much meaning as types -- wasDerivedFrom and dependsOn are not interchangeable -- and a conformance profile that asks whether types are grounded has to be able to ask it of edges too.

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

class SetEdgeSemanticsOp(ConfigBaseModel):
    """Ground edge relations in an external vocabulary.

    The vertex op's counterpart. Relations carry as much meaning as types --
    ``wasDerivedFrom`` and ``dependsOn`` are not interchangeable -- and a
    conformance profile that asks whether types are grounded has to be able to
    ask it of edges too.
    """

    op: Literal["set_edge_semantics"] = "set_edge_semantics"
    edges: list[EdgeSelector] = PydanticField(
        ...,
        description="Edge triples whose grounding changes.",
        min_length=1,
    )
    semantics: Semantics | None = PydanticField(
        default=None,
        description=(
            "Grounding applied to every selected edge; ``None`` clears it. Not "
            "``FieldSemantics``: a unit on an edge is meaningless, and the model "
            "split is what makes ``unit:`` here a validation error."
        ),
    )

    @model_validator(mode="after")
    def _validate_unique_selectors(self) -> SetEdgeSemanticsOp:
        _validate_unique_edge_selectors(self.edges, kind="set_edge_semantics")
        return self

Attributes

edges = PydanticField(..., description='Edge triples whose grounding changes.', min_length=1) class-attribute instance-attribute

op = 'set_edge_semantics' class-attribute instance-attribute

semantics = PydanticField(default=None, description='Grounding applied to every selected edge; ``None`` clears it. Not ``FieldSemantics``: a unit on an edge is meaningless, and the model split is what makes ``unit:`` here a validation error.') class-attribute instance-attribute

SetFieldSemanticsOp

Bases: ConfigBaseModel

Ground vertex and edge properties, including their unit of measure.

Takes :class:~graflo.architecture.schema.semantics.FieldSemantics rather than :class:~graflo.architecture.schema.semantics.Semantics, which is the entire reason this is a third op rather than a mode of the vertex one: only a property may carry unit, and the two models are kept apart so that unit: on a type is a validation error rather than a silent no-op.

A target names either a vertex property (vertex + field) or an edge property (source / target / relation + field).

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

class SetFieldSemanticsOp(ConfigBaseModel):
    """Ground vertex and edge properties, including their unit of measure.

    Takes :class:`~graflo.architecture.schema.semantics.FieldSemantics` rather
    than :class:`~graflo.architecture.schema.semantics.Semantics`, which is the
    entire reason this is a third op rather than a mode of the vertex one: only
    a property may carry ``unit``, and the two models are kept apart so that
    ``unit:`` on a type is a validation error rather than a silent no-op.

    A target names either a vertex property (``vertex`` + ``field``) or an
    edge property (``source`` / ``target`` / ``relation`` + ``field``).
    """

    op: Literal["set_field_semantics"] = "set_field_semantics"
    targets: list[FieldSemanticsTarget | EdgeFieldSemanticsTarget] = PydanticField(
        ...,
        description="Properties whose grounding changes.",
        min_length=1,
    )

    @model_validator(mode="after")
    def _validate_unique_targets(self) -> SetFieldSemanticsOp:
        keys = [t.key() for t in self.targets]
        if len(keys) != len(set(keys)):
            raise ValueError(
                "set_field_semantics targets must be unique by (vertex, field) or "
                "(source, target, relation, field)"
            )
        return self

Attributes

op = 'set_field_semantics' class-attribute instance-attribute

targets = PydanticField(..., description='Properties whose grounding changes.', min_length=1) class-attribute instance-attribute

SetInverseEmissionOp

Bases: ConfigBaseModel

Set or clear emit_inverse on edge steps, addressed by position.

The ingestion half of a materialized inverse, as a primitive: which steps mirror the edges they write into the declared inverse. A step is addressed by resource and :class:EdgeStepRef (at / step / link), so the op says exactly which steps change and nothing is inferred at replay time.

Enabling is refused where the flag could never write anything -- a step naming exactly one edge whose relation has no declared pair, is symmetric, or has no declared inverse edge. A step whose relation comes from the data is accepted; it mirrors per document what has a materialized inverse.

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

class SetInverseEmissionOp(ConfigBaseModel):
    """Set or clear ``emit_inverse`` on edge steps, addressed by position.

    The ingestion half of a materialized inverse, as a primitive: which steps
    mirror the edges they write into the declared inverse. A step is addressed
    by resource and :class:`EdgeStepRef` (``at`` / ``step`` / ``link``), so the
    op says exactly which steps change and nothing is inferred at replay time.

    Enabling is refused where the flag could never write anything -- a step
    naming exactly one edge whose relation has no declared pair, is symmetric,
    or has no declared inverse edge. A step whose relation comes from the data
    is accepted; it mirrors per document what has a materialized inverse.
    """

    op: Literal["set_inverse_emission"] = "set_inverse_emission"
    steps: dict[str, list[EdgeStepRef]] = PydanticField(
        ...,
        description="Edge steps whose flag changes: ``{resource: [step ref, ...]}``.",
        min_length=1,
    )
    enabled: bool = PydanticField(
        default=True,
        description="``False`` clears the flag.",
    )

    @model_validator(mode="after")
    def _validate_steps(self) -> SetInverseEmissionOp:
        empty = sorted(name for name, refs in self.steps.items() if not refs)
        if empty:
            raise ValueError(f"set_inverse_emission: no steps given for {empty}")
        for name, refs in self.steps.items():
            keys = [ref.sort_key for ref in refs]
            if len(set(keys)) != len(keys):
                raise ValueError(
                    f"set_inverse_emission: steps of {name!r} must be unique"
                )
        return self

Attributes

enabled = PydanticField(default=True, description='``False`` clears the flag.') class-attribute instance-attribute

op = 'set_inverse_emission' class-attribute instance-attribute

steps = PydanticField(..., description='Edge steps whose flag changes: ``{resource: [step ref, ...]}``.', min_length=1) class-attribute instance-attribute

SetNativeInversesOp

Bases: ConfigBaseModel

Have the database maintain declared inverses (TigerGraph WITH REVERSE_EDGE).

Physical, not logical: adds relations to, or removes them from, db_profile.native_inverses. Keyed by relation, as TigerGraph is: the reverse type belongs to the edge type, which spans every (S, T) pair of the relation. The reverse type is named by the declared pair, so the pair must be declared (:class:DeclareEdgeInversesOp). Refused for symmetric relations, where explicit inverse edges exist, on undirected edges, for a relation stored under several physical names, and on non-TigerGraph profiles.

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

class SetNativeInversesOp(ConfigBaseModel):
    """Have the database maintain declared inverses (TigerGraph ``WITH REVERSE_EDGE``).

    Physical, not logical: adds relations to, or removes them from,
    ``db_profile.native_inverses``. Keyed by relation, as TigerGraph is: the
    reverse type belongs to the edge type, which spans every ``(S, T)`` pair of
    the relation. The reverse type is named by the declared pair, so the pair
    must be declared (:class:`DeclareEdgeInversesOp`). Refused for symmetric
    relations, where explicit inverse edges exist, on undirected edges, for a
    relation stored under several physical names, and on non-TigerGraph
    profiles.
    """

    op: Literal["set_native_inverses"] = "set_native_inverses"
    relations: list[str] = PydanticField(
        ...,
        description="Paired relations whose declared inverse the database maintains.",
        min_length=1,
    )
    enabled: bool = PydanticField(
        default=True,
        description="``False`` withdraws the native inverse.",
    )

    @model_validator(mode="after")
    def _validate_unique(self) -> SetNativeInversesOp:
        if len(set(self.relations)) != len(self.relations):
            raise ValueError("set_native_inverses: relations must be unique")
        return self

Attributes

enabled = PydanticField(default=True, description='``False`` withdraws the native inverse.') class-attribute instance-attribute

op = 'set_native_inverses' class-attribute instance-attribute

relations = PydanticField(..., description='Paired relations whose declared inverse the database maintains.', min_length=1) class-attribute instance-attribute

SetVertexDescriptionsOp

Bases: ConfigBaseModel

Set or clear the human-readable description of existing vertex types.

A description was authorable only when a type was first written, so a change to one -- a merge folding two types together and joining what each side said about it, or a correction to what an inferred type means -- had no operation and could only be diffed as inexpressible. Like grounding, a description is never consulted at execution time: this op cannot change how anything ingests or stores.

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

class SetVertexDescriptionsOp(ConfigBaseModel):
    """Set or clear the human-readable description of existing vertex types.

    A description was authorable only when a type was first written, so a
    change to one -- a merge folding two types together and joining what each
    side said about it, or a correction to what an inferred type means -- had no
    operation and could only be diffed as inexpressible. Like grounding, a
    description is never consulted at execution time: this op cannot change how
    anything ingests or stores.
    """

    op: Literal["set_vertex_descriptions"] = "set_vertex_descriptions"
    descriptions: dict[str, str | None] = PydanticField(
        ...,
        description=(
            "Per-vertex description: ``{vertex_name: text}``. ``None`` clears it, "
            "which is what makes the op invertible."
        ),
        min_length=1,
    )

Attributes

descriptions = PydanticField(..., description='Per-vertex description: ``{vertex_name: text}``. ``None`` clears it, which is what makes the op invertible.', min_length=1) class-attribute instance-attribute

op = 'set_vertex_descriptions' class-attribute instance-attribute

SetVertexSemanticsOp

Bases: ConfigBaseModel

Ground vertex types in an external vocabulary.

Semantics were authorable only when a type was first written: no operation could attach an iri to a type that already existed, so a manifest that arrived ungrounded — an inferred one, or anything predating the block — could never be grounded through the op system, only rewritten by hand.

Grounding is purely additive and never consulted at execution time, so this op cannot change how anything ingests or stores. What it changes is whether a reader who did not author the schema can tell what a type denotes.

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

class SetVertexSemanticsOp(ConfigBaseModel):
    """Ground vertex types in an external vocabulary.

    Semantics were authorable only when a type was first written: no operation
    could attach an ``iri`` to a type that already existed, so a manifest that
    arrived ungrounded — an inferred one, or anything predating the block — could
    never be grounded through the op system, only rewritten by hand.

    Grounding is purely additive and never consulted at execution time, so this
    op cannot change how anything ingests or stores. What it changes is whether a
    reader who did not author the schema can tell what a type denotes.
    """

    op: Literal["set_vertex_semantics"] = "set_vertex_semantics"
    semantics: dict[str, Semantics | None] = PydanticField(
        ...,
        description=(
            "Per-vertex grounding: ``{vertex_name: Semantics}``. ``None`` clears "
            "the block, which is what makes the op invertible."
        ),
        min_length=1,
    )

Attributes

op = 'set_vertex_semantics' class-attribute instance-attribute

semantics = PydanticField(..., description='Per-vertex grounding: ``{vertex_name: Semantics}``. ``None`` clears the block, which is what makes the op invertible.', min_length=1) class-attribute instance-attribute

SharedDerivation

Bases: ConfigBaseModel

One derivation shared by several members, varying only in parameters.

The compact spelling of the member-keyed form for the common case: the call is the same for every member and only a parameter changes — a marker prefix per class — or nothing does. members is a list of member classes, or a dict from member to the parameters that differ; each member's derivation is spec with those parameters laid over spec.params. Anything else that differs between members — the input columns, the function — is a different derivation: spell it with the explicit {member: spec} dict.

Expands to that dict; the lowering never sees this model.

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

class SharedDerivation(ConfigBaseModel):
    """One derivation shared by several members, varying only in parameters.

    The compact spelling of the member-keyed form for the common case: the
    call is the same for every member and only a parameter changes — a marker
    prefix per class — or nothing does. ``members`` is a list of member
    classes, or a dict from member to the parameters that differ; each
    member's derivation is ``spec`` with those parameters laid over
    ``spec.params``. Anything else that differs between members — the input
    columns, the function — is a different derivation: spell it with the
    explicit ``{member: spec}`` dict.

    Expands to that dict; the lowering never sees this model.
    """

    spec: DerivationSpec = PydanticField(
        ...,
        description="The derivation every member shares.",
    )
    members: list[str] | dict[str, dict[str, Any]] = PydanticField(
        ...,
        description=(
            "Member classes sharing ``spec``: a list when nothing varies, or "
            "``{member: {param: value}}`` naming what does."
        ),
    )

    @model_validator(mode="after")
    def _validate_members(self) -> SharedDerivation:
        if not self.members:
            raise ValueError("SharedDerivation: members must name at least one class")
        if isinstance(self.members, list) and len(set(self.members)) != len(
            self.members
        ):
            raise ValueError("SharedDerivation: members lists a class twice")
        return self

    def expand(self) -> dict[str, DerivationSpec]:
        """The explicit ``{member: spec}`` dict this stands for."""
        if isinstance(self.members, list):
            return {member: self.spec for member in self.members}
        return {
            member: self.spec.model_copy(
                update={"params": {**self.spec.params, **overrides}}
            )
            for member, overrides in self.members.items()
        }

Attributes

members = PydanticField(..., description='Member classes sharing ``spec``: a list when nothing varies, or ``{member: {param: value}}`` naming what does.') class-attribute instance-attribute

spec = PydanticField(..., description='The derivation every member shares.') class-attribute instance-attribute

Methods:

expand()

The explicit {member: spec} dict this stands for.

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

def expand(self) -> dict[str, DerivationSpec]:
    """The explicit ``{member: spec}`` dict this stands for."""
    if isinstance(self.members, list):
        return {member: self.spec for member in self.members}
    return {
        member: self.spec.model_copy(
            update={"params": {**self.spec.params, **overrides}}
        )
        for member, overrides in self.members.items()
    }

SideIdentity

Bases: ConfigBaseModel

Per-side/per-member shorthand for a cluster's merged identity funnel.

Each entry is one funnel branch: a single canonical attribute, or an ordered composite (list[str]). left / right supply the default branch chain for every member declared on that side; members overrides it for specific member classes, keyed by the member's own or canonical name. Every chain is merged into one global branch order — see :func:~graflo.architecture.evolution.merge.side_identity_to_funnel — so declaring the same relative order on every member is required; two members disagreeing on the order of two branches raises.

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

class SideIdentity(ConfigBaseModel):
    """Per-side/per-member shorthand for a cluster's merged identity funnel.

    Each entry is one funnel branch: a single canonical attribute, or an
    ordered composite (``list[str]``). ``left`` / ``right`` supply the default
    branch chain for every member declared on that side; ``members`` overrides
    it for specific member classes, keyed by the member's own or canonical
    name. Every chain is merged into one global branch order — see
    :func:`~graflo.architecture.evolution.merge.side_identity_to_funnel` —
    so declaring the same relative order on every member is required; two
    members disagreeing on the order of two branches raises.
    """

    left: list[IdentityBranchSpec] | None = PydanticField(
        default=None,
        description="Default ordered branch chain for every left member.",
    )
    right: list[IdentityBranchSpec] | None = PydanticField(
        default=None,
        description="Default ordered branch chain for every right member.",
    )
    members: dict[str, list[IdentityBranchSpec]] = PydanticField(
        default_factory=dict,
        description="Per-member branch chain, overriding the side default.",
    )

    @model_validator(mode="after")
    def _validate_shape(self) -> SideIdentity:
        if self.left is None and self.right is None and not self.members:
            raise ValueError(
                "SideIdentity requires at least one of left, right, or members"
            )
        return self

Attributes

left = PydanticField(default=None, description='Default ordered branch chain for every left member.') class-attribute instance-attribute

members = PydanticField(default_factory=dict, description='Per-member branch chain, overriding the side default.') class-attribute instance-attribute

right = PydanticField(default=None, description='Default ordered branch chain for every right member.') class-attribute instance-attribute

VertexEquivalence

Bases: ConfigBaseModel

Collapse one or more left classes and one or more right classes into one.

GraFlo applies this map deterministically; it does not infer semantic matches. left / right accept a bare class name (a 1-1 equivalence) or a list (an n-ary cluster): {Company, Shop} ~ {Org, Branch} -> Company. Declaring more than one member on a side is a merge and requires MergeManifestsOp.allow_merges=True.

Properties with the same spelling on every member after alignment fuse by exact name without an entry in properties — list only renames and identity-flagged fields.

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

class VertexEquivalence(ConfigBaseModel):
    """Collapse one or more left classes and one or more right classes into one.

    GraFlo applies this map deterministically; it does not infer semantic
    matches. ``left`` / ``right`` accept a bare class name (a 1-1 equivalence)
    or a list (an n-ary cluster): ``{Company, Shop} ~ {Org, Branch} ->
    Company``. Declaring more than one member on a side is a merge and
    requires ``MergeManifestsOp.allow_merges=True``.

    Properties with the same spelling on every member after alignment fuse by
    exact name without an entry in ``properties`` — list only renames and
    identity-flagged fields.
    """

    left: str | list[str] = PydanticField(
        ..., description="One or more left-manifest vertex type names."
    )
    right: str | list[str] = PydanticField(
        ..., description="One or more right-manifest vertex type names."
    )
    into: str | None = PydanticField(
        default=None,
        description=(
            "Merged vertex type name (may equal a member's name, or be a "
            "new name). Omitted, the name comes from the canonical map that "
            "maps a member, or from the one spelling every member shares."
        ),
    )
    properties: list[PropertyEquivalence] = PydanticField(
        default_factory=list,
        description="Property alignment map applied before the vertex merge.",
    )
    identity: list[str] | IdentityFunnel | SideIdentity | None = PydanticField(
        default=None,
        description=(
            "Optional explicit merged identity, in canonical attribute names "
            "(after alignment): a natural key, an explicit funnel, or a "
            "`SideIdentity` shorthand lowered to one funnel. When unset, "
            "identity is carried through only if every member agrees after "
            "alignment (plus any `PropertyEquivalence.identity` flags); "
            "disagreement with nothing declared raises `MergeIdentityError`."
        ),
    )
    retire: Literal["demote", "keep"] = PydanticField(
        default="demote",
        description=(
            "What becomes of each member's pre-merge identity fields when "
            "`identity` is declared. `demote` keeps them as lookup-only "
            "secondary identities on `into`; `keep` drops them. Unused when "
            "`identity` is unset."
        ),
    )

    @property
    def left_members(self) -> list[str]:
        return _member_list(self.left)

    @property
    def right_members(self) -> list[str]:
        return _member_list(self.right)

    def members(self, side: Literal["left", "right"]) -> list[str]:
        return self.left_members if side == "left" else self.right_members

    def property_maps(
        self, side: Literal["left", "right"]
    ) -> dict[str, dict[str, str]]:
        """``{member: {old_field: into_field}}`` for *side*, bare strings expanded."""
        member_names = self.members(side)
        out: dict[str, dict[str, str]] = {}
        for pe in self.properties:
            spec = pe.left if side == "left" else pe.right
            if spec is None:
                continue
            per_member = (
                dict.fromkeys(member_names, spec) if isinstance(spec, str) else spec
            )
            for member, old in per_member.items():
                if old == pe.into:
                    continue
                bucket = out.setdefault(member, {})
                existing = bucket.get(old)
                if existing is not None and existing != pe.into:
                    raise ValueError(
                        f"VertexEquivalence: {side}:{member}.{old!r} would rename "
                        f"to both {existing!r} and {pe.into!r}"
                    )
                bucket[old] = pe.into
        return out

    @model_validator(mode="after")
    def _validate_members(self) -> VertexEquivalence:
        for side, members in (
            ("left", self.left_members),
            ("right", self.right_members),
        ):
            if not members:
                raise ValueError(
                    f"VertexEquivalence: {side} must name at least one class"
                )
            if len(members) != len(set(members)):
                raise ValueError(
                    f"VertexEquivalence: {side} lists a class more than once: {members}"
                )
        for pe in self.properties:
            for side, spec in (("left", pe.left), ("right", pe.right)):
                if isinstance(spec, dict):
                    unknown = sorted(set(spec) - set(self.members(side)))
                    if unknown:
                        raise ValueError(
                            f"VertexEquivalence: property equivalence into "
                            f"{pe.into!r} names {side} member(s) {unknown} not "
                            f"in {side}={self.members(side)}"
                        )
        if self.identity is not None:
            flagged = [pe.into for pe in self.properties if pe.identity]
            if flagged:
                raise ValueError(
                    "VertexEquivalence: `identity` is declared on the cluster; "
                    f"PropertyEquivalence.identity=True on {flagged} is "
                    "redundant and conflicting — declare the merged key one "
                    "way, not both"
                )
        return self

Attributes

identity = PydanticField(default=None, description='Optional explicit merged identity, in canonical attribute names (after alignment): a natural key, an explicit funnel, or a `SideIdentity` shorthand lowered to one funnel. When unset, identity is carried through only if every member agrees after alignment (plus any `PropertyEquivalence.identity` flags); disagreement with nothing declared raises `MergeIdentityError`.') class-attribute instance-attribute

into = PydanticField(default=None, description="Merged vertex type name (may equal a member's name, or be a new name). Omitted, the name comes from the canonical map that maps a member, or from the one spelling every member shares.") class-attribute instance-attribute

left = PydanticField(..., description='One or more left-manifest vertex type names.') class-attribute instance-attribute

left_members property

properties = PydanticField(default_factory=list, description='Property alignment map applied before the vertex merge.') class-attribute instance-attribute

retire = PydanticField(default='demote', description="What becomes of each member's pre-merge identity fields when `identity` is declared. `demote` keeps them as lookup-only secondary identities on `into`; `keep` drops them. Unused when `identity` is unset.") class-attribute instance-attribute

right = PydanticField(..., description='One or more right-manifest vertex type names.') class-attribute instance-attribute

right_members property

Methods:

members(side)

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

def members(self, side: Literal["left", "right"]) -> list[str]:
    return self.left_members if side == "left" else self.right_members

property_maps(side)

{member: {old_field: into_field}} for side, bare strings expanded.

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

def property_maps(
    self, side: Literal["left", "right"]
) -> dict[str, dict[str, str]]:
    """``{member: {old_field: into_field}}`` for *side*, bare strings expanded."""
    member_names = self.members(side)
    out: dict[str, dict[str, str]] = {}
    for pe in self.properties:
        spec = pe.left if side == "left" else pe.right
        if spec is None:
            continue
        per_member = (
            dict.fromkeys(member_names, spec) if isinstance(spec, str) else spec
        )
        for member, old in per_member.items():
            if old == pe.into:
                continue
            bucket = out.setdefault(member, {})
            existing = bucket.get(old)
            if existing is not None and existing != pe.into:
                raise ValueError(
                    f"VertexEquivalence: {side}:{member}.{old!r} would rename "
                    f"to both {existing!r} and {pe.into!r}"
                )
            bucket[old] = pe.into
    return out

Functions:

ops_reaching_ingestion(ops)

Names of ops whose effect extends into ingestion_model, in order.

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

def ops_reaching_ingestion(ops: Sequence[Any]) -> list[str]:
    """Names of *ops* whose effect extends into ``ingestion_model``, in order."""
    return [
        name
        for name in (getattr(op, "op", None) for op in ops)
        if isinstance(name, str) and name in INGESTION_REWRITING_OPS
    ]

validate_merge_sources(sources, into, *, kind)

Reject a merge whose sources repeat or include the target.

The docstrings promise both; enforcing them at parse time means a serialized change set fails where it is read rather than where it is replayed, after the ops before it have already been applied.

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

def validate_merge_sources(sources: Sequence[str], into: str, *, kind: str) -> None:
    """Reject a merge whose sources repeat or include the target.

    The docstrings promise both; enforcing them at parse time means a
    serialized change set fails where it is read rather than where it is
    replayed, after the ops before it have already been applied.
    """
    if len(set(sources)) != len(sources):
        raise ValueError(f"{kind}: sources list a name more than once: {list(sources)}")
    if into in sources:
        raise ValueError(f"{kind}: `into` {into!r} must not appear in `sources`")

validate_rename_map_is_injective(renames, *, kind, merge_hint)

Reject a rename map that would collapse two names onto one.

A rename is a relabelling: it must not change how many types exist. Two sources sharing a target is a merge, and the merge ops exist precisely because merging needs decisions a rename cannot express — which properties survive, how identity combines, what happens to edges that become self-loops. Left unchecked the collapse is silent: the name-keyed lookup maps in VertexConfig / EdgeConfig keep the last definition and the earlier one is shadowed but still serialized.

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

def validate_rename_map_is_injective(
    renames: dict[str, str], *, kind: str, merge_hint: str
) -> None:
    """Reject a rename map that would collapse two names onto one.

    A rename is a relabelling: it must not change how many types exist. Two sources
    sharing a target is a *merge*, and the merge ops exist precisely because merging
    needs decisions a rename cannot express — which properties survive, how identity
    combines, what happens to edges that become self-loops. Left unchecked the
    collapse is silent: the name-keyed lookup maps in ``VertexConfig`` / ``EdgeConfig``
    keep the last definition and the earlier one is shadowed but still serialized.
    """
    collisions: dict[str, list[str]] = {}
    for source, target in renames.items():
        collisions.setdefault(target, []).append(source)
    collapsed = {
        target: sorted(sources)
        for target, sources in collisions.items()
        if len(sources) > 1
    }
    if collapsed:
        detail = "; ".join(
            f"{target!r} is the target of {sources}"
            for target, sources in sorted(collapsed.items())
        )
        raise ValueError(
            f"{kind} rename map is not injective: {detail}. "
            f"Renaming cannot merge types — use {merge_hint}."
        )

validate_vocabulary_is_idempotent(mapping, *, kind)

Reject a chain or a swap: a canonical vocabulary has fixed points.

A name that is both a source that moves and a target of another entry ({X: Z, Z: Q}, {A: B, B: A}) makes the map non-idempotent — applying it twice is not applying it once — so it cannot be read as a vocabulary, where a canonical name is by definition one nothing maps away from. Such a map is a relabel, which :class:CanonicalizeOp expresses (simultaneous application over the original schema).

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

def validate_vocabulary_is_idempotent(mapping: Mapping[str, str], *, kind: str) -> None:
    """Reject a chain or a swap: a canonical vocabulary has fixed points.

    A name that is both a source that moves and a target of another entry
    (``{X: Z, Z: Q}``, ``{A: B, B: A}``) makes the map non-idempotent — applying
    it twice is not applying it once — so it cannot be read as a vocabulary,
    where a canonical name is by definition one nothing maps away from. Such a
    map is a *relabel*, which :class:`CanonicalizeOp` expresses (simultaneous
    application over the original schema).
    """
    moving = {source for source, target in mapping.items() if source != target}
    targets = {target for source, target in mapping.items() if source != target}
    chained = sorted(moving & targets)
    if chained:
        raise ValueError(
            f"{kind}: {chained} are both a source that moves and a canonical "
            "target. A canonical vocabulary has fixed points — a chain or a "
            "swap is a relabel, which CanonicalizeOp expresses."
        )

validate_vocabulary_map(vertices, relations, properties, *, allow_merges, kind, merge_hint)

Reject an unacknowledged collapse in a vocabulary map.

A vocabulary map is a function on names, so its groups are its fibers: every source of one target, including a self entry t: t that declares an existing t a member of its own group. A group of more than one name is a merge — it fuses entities and can create self-relations — so it must be acknowledged rather than inferred from the map. Per-class attribute maps are plain renames and must be injective outright.

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

def validate_vocabulary_map(
    vertices: Mapping[str, str],
    relations: Mapping[str, str],
    properties: Mapping[str, Mapping[str, str]],
    *,
    allow_merges: bool,
    kind: str,
    merge_hint: str,
) -> None:
    """Reject an unacknowledged collapse in a vocabulary map.

    A vocabulary map is a function on names, so its groups are its fibers:
    every source of one target, **including** a self entry ``t: t`` that
    declares an existing ``t`` a member of its own group. A group of more than
    one name is a merge — it fuses entities and can create self-relations — so
    it must be acknowledged rather than inferred from the map. Per-class
    attribute maps are plain renames and must be injective outright.
    """
    if not allow_merges:
        for noun, mapping in (("class", vertices), ("relation", relations)):
            collapsed = {
                target: sorted(sources)
                for target, sources in vocabulary_groups(mapping).items()
                if len(sources) > 1
            }
            if collapsed:
                detail = "; ".join(
                    f"{target!r} is the target of {sources}"
                    for target, sources in sorted(collapsed.items())
                )
                raise ValueError(
                    f"{kind}: {noun} map collapses names: {detail}. A merge is a "
                    f"stated intent — {merge_hint}."
                )
    for source_class, attr_map in properties.items():
        validate_rename_map_is_injective(
            dict(attr_map),
            kind=f"{kind} property (class {source_class!r})",
            merge_hint="a transform that combines the fields upstream",
        )

vocabulary_groups(mapping)

{target: [sources]} — the fibers of a vocabulary map, self entries included.

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

def vocabulary_groups(mapping: Mapping[str, str]) -> dict[str, list[str]]:
    """``{target: [sources]}`` — the fibers of a vocabulary map, self entries included."""
    groups: dict[str, list[str]] = {}
    for source, target in mapping.items():
        groups.setdefault(target, []).append(source)
    return groups