graflo.architecture.evolution.ops

Typed manifest evolution operations.

AddEdgePropertiesOp

Bases: ConfigBaseModel

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

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

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

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

AddInverseEdgesOp

Bases: ConfigBaseModel

Add inverse edge relations for matching relations across schema and ingestion.

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

class AddInverseEdgesOp(ConfigBaseModel):
    """Add inverse edge relations for matching relations across schema and ingestion."""

    op: Literal["add_inverse_edges"] = "add_inverse_edges"
    relations: dict[str, str] = PydanticField(
        ...,
        description=(
            "Relation inverse map: ``{relation_name: inverse_relation_name}``."
        ),
    )

AddVertexPropertiesOp

Bases: ConfigBaseModel

Add vertex properties to existing logical vertex types.

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

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

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

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

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.",
    )

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."
        ),
    )

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 auto-kept; list them explicitly in keep_edges when needed.

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

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** auto-kept; list them explicitly
    in ``keep_edges`` when needed.

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

    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.",
    )
    keep_resources: list[str] | None = PydanticField(
        default=None,
        description="Optional ingestion resource names to retain after graph slice.",
    )
    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)"
                )
        return self

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, ...]}``."
        ),
    )

RemoveEdgesOp

Bases: ConfigBaseModel

Remove logical edge relations from schema, profile, and ingestion selectors.

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

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

    op: Literal["remove_edges"] = "remove_edges"
    relations: list[str] = PydanticField(
        ...,
        description="Relation names to remove from edge definitions and references.",
        min_length=1,
    )

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, ...]}``."
        ),
    )

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,
    )

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}}``."
        ),
    )

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"
    relations: dict[str, str] = PydanticField(
        ...,
        description="Relation rename map: ``{old_relation: new_relation}``.",
    )

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"
    resources: dict[str, str] = PydanticField(
        ...,
        description="Ingestion resource rename map: ``{old_resource: new_resource}``.",
    )

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}}``."
        ),
    )

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"
    vertices: dict[str, str] = PydanticField(
        ...,
        description="Vertex rename map: ``{old_vertex: new_vertex}``.",
    )

SanitizeOp

Bases: ConfigBaseModel

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

Composes (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.

    Composes (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."
        ),
    )