graflo.architecture.evolution.ops.identities

Ops that replace or extend vertex and edge identities.

Attributes

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

Classes

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/identities.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

AssignedIdentityTarget

Bases: ConfigBaseModel

Target an assigned identity: an intentional UUID primary key.

Source code in graflo/architecture/evolution/ops/identities.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/identities.py

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

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

Attributes

mode = 'blank' class-attribute instance-attribute

EdgeIdentitiesEntry

Bases: ConfigBaseModel

New uniqueness keys for one edge triple.

Source code in graflo/architecture/evolution/ops/identities.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/identities.py

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

FunnelIdentityTarget

Bases: ConfigBaseModel

Target an identity funnel: fallback branches digested into digest_field.

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/identities.py

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

    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.",
    )
    digest_field: str = PydanticField(
        default="id",
        min_length=1,
        description="Property the digest is stored in; the vertex's identity field.",
    )

    @model_validator(mode="after")
    def _digest_field_is_no_branch_field(self) -> FunnelIdentityTarget:
        # The cast drops a digest vertex's identity field from each record, so
        # a branch reading the same field could never complete.
        if self.digest_field in self.funnel.field_names:
            raise ValueError(
                f"funnel identity: digest_field {self.digest_field!r} is also a "
                "branch field; the digest would replace that branch's input"
            )
        return self

Attributes

digest_field = PydanticField(default='id', min_length=1, description="Property the digest is stored in; the vertex's identity field.") class-attribute instance-attribute

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/identities.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

IdentityReplacement

Bases: ConfigBaseModel

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

Source code in graflo/architecture/evolution/ops/identities.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

NaturalIdentityTarget

Bases: ConfigBaseModel

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

Source code in graflo/architecture/evolution/ops/identities.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

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/identities.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

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/identities.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).

Source code in graflo/architecture/evolution/ops/identities.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).
    """

    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