graflo.architecture.evolution.rewrite

Structured rewrite of vertex names in pipeline dicts and related resource fields.

Attributes

logger = logging.getLogger(__name__) module-attribute

Functions:

cast_step(x)

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

def cast_step(x: Any) -> dict[str, Any]:
    if not isinstance(x, dict):
        raise TypeError(f"expected dict step, got {type(x)}")
    return x

close_routers_in_pipeline(pipeline, vocabulary, *, declared_properties=None, pass_through=None)

pipeline with every open router closed over vocabulary.

An open router passes a value missing from its type_map through as the class name. Closing it lists each class of vocabulary the table does not already key -- as itself -- and sets type_map_only, so it routes exactly what it routed when vocabulary was the whole schema. A router with vertex_types gets entries only for the classes it lists. Entries already there, authored or written by a rename, are kept. Steps keep their authored spelling, at every level.

A class the router reached only by pass-through took the part of the router-level from it declares; a listed class must declare all of it. Given declared_properties (property names per class), every class such a value routes to after closing that lacks a target gets that declared part as its vertex_from_map entry, so closing never refuses a projection the open router accepted. pass_through names those values when the table was rewritten since (see :func:pass_through_classes); by default they are the values closing itself lists.

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

def close_routers_in_pipeline(
    pipeline: list[dict[str, Any]],
    vocabulary: Collection[str],
    *,
    declared_properties: Mapping[str, Collection[str]] | None = None,
    pass_through: Collection[str] | None = None,
) -> list[dict[str, Any]]:
    """*pipeline* with every open router closed over *vocabulary*.

    An open router passes a value missing from its ``type_map`` through as the
    class name. Closing it lists each class of *vocabulary* the table does not
    already key -- as itself -- and sets ``type_map_only``, so it routes
    exactly what it routed when *vocabulary* was the whole schema. A router
    with ``vertex_types`` gets entries only for the classes it lists. Entries
    already there, authored or written by a rename, are kept. Steps keep their
    authored spelling, at every level.

    A class the router reached only by pass-through took the part of the
    router-level ``from`` it declares; a listed class must declare all of it.
    Given *declared_properties* (property names per class), every class such
    a value routes to after closing that lacks a target gets that declared
    part as its ``vertex_from_map`` entry, so closing never refuses a
    projection the open router accepted. *pass_through* names those values
    when the table was rewritten since (see :func:`pass_through_classes`);
    by default they are the values closing itself lists.
    """
    out = deepcopy(pipeline)
    _close_routers(out, sorted(vocabulary), declared_properties, pass_through)
    return out

collect_endpoint_selectors(pipeline)

(vertex_name, selector) for every use of a secondary identity.

Each edge endpoint matched on one, and each class a vertex step or a router finds by one (find). Endpoints resolving through the primary identity are omitted — they carry no dependency on a named secondary identity.

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

def collect_endpoint_selectors(
    pipeline: list[dict[str, Any]],
) -> list[tuple[str, str | list[str]]]:
    """``(vertex_name, selector)`` for every use of a secondary identity.

    Each edge endpoint matched on one, and each class a vertex step or a
    router finds by one (``find``). Endpoints resolving through the primary
    identity are omitted — they carry no dependency on a named secondary
    identity.
    """
    out: list[tuple[str, str | list[str]]] = []
    _collect_endpoint_selectors_in_step(pipeline, out)
    return out

edge_payloads(step)

The edge payloads step carries, as mutable references into it.

An edge step is spelled three ways: nested under edge or create_edge, or flat, with the endpoints on the step itself ({source, target} / {from, to}, or type: edge). A walker that only looks under the two keys silently skips every flat step -- the form most hand-written pipelines use.

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

def edge_payloads(step: dict[str, Any]) -> list[dict[str, Any]]:
    """The edge payloads *step* carries, as mutable references into it.

    An edge step is spelled three ways: nested under ``edge`` or
    ``create_edge``, or flat, with the endpoints on the step itself
    (``{source, target}`` / ``{from, to}``, or ``type: edge``). A walker that
    only looks under the two keys silently skips every flat step -- the form
    most hand-written pipelines use.
    """
    payloads = [
        payload
        for key in ("edge", "create_edge")
        if isinstance(payload := step.get(key), dict)
    ]
    if payloads or "vertex" in step:
        return payloads
    if step.get("type") == "edge" or (
        ("source" in step or "from" in step) and ("target" in step or "to" in step)
    ):
        return [step]
    return []

evolve_router(payload, classes, *, declared=None)

The routers that replace a router once an op maps its classes by classes.

classes maps an old class to its new name, or to None when the op removes it; a class it does not name is unchanged. declared is the schema's classes before the op. Every discriminator value must route as before, to its class renamed (or nowhere, if it was removed): the table's values and vertex_types are renamed, and a pass-through router gains {old: new} for each renamed class it reaches. A value that named no class may come to route to a new name -- as an open router always routes a declared name -- but only when the router reached every class renamed onto it.

Where those fields cannot say that -- a vertex_types router reaching one class of a merge and not another, or a removed table entry whose key names a class -- the router is closed instead: type_map_only over exactly the values it accepted, then renamed.

A merge of classes the router projects differently -- each its own vertex_from_map entry, or the shared from -- cannot keep one projection per class: the router is split into closed routers on the same type_field and role, one per projection, each keyed by the values that reached its classes. The values stay disjoint, so no row is routed twice. Without declared the router's reach is unknown; it is neither closed nor split, and colliding projections are unioned.

Without declared, a new name is taken to be new to the schema and every other name to be declared.

Returns the replacing payloads: one as a rule, none when the router routes nothing any more, several when it was split.

Raises:

Type	Description
ValueError	If a merge joins a class the router only looks up with one it writes; one class cannot be both.

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

def evolve_router(
    payload: dict[str, Any],
    classes: Mapping[str, str | None],
    *,
    declared: Collection[str] | None = None,
) -> list[dict[str, Any]]:
    """The routers that replace a router once an op maps its classes by *classes*.

    *classes* maps an old class to its new name, or to ``None`` when the op
    removes it; a class it does not name is unchanged. *declared* is the
    schema's classes before the op. Every discriminator value must route as
    before, to its class renamed (or nowhere, if it was removed): the table's
    values and ``vertex_types`` are renamed, and a pass-through router gains
    ``{old: new}`` for each renamed class it reaches. A value that named no
    class may come to route to a new name -- as an open router always routes
    a declared name -- but only when the router reached every class renamed
    onto it.

    Where those fields cannot say that -- a ``vertex_types`` router reaching
    one class of a merge and not another, or a removed table entry whose key
    names a class -- the router is closed instead: ``type_map_only`` over
    exactly the values it accepted, then renamed.

    A merge of classes the router projects differently -- each its own
    ``vertex_from_map`` entry, or the shared ``from`` -- cannot keep one
    projection per class: the router is split into closed routers on the same
    ``type_field`` and ``role``, one per projection, each keyed by the values
    that reached its classes. The values stay disjoint, so no row is routed
    twice. Without *declared* the router's reach is unknown; it is neither
    closed nor split, and colliding projections are unioned.

    Without *declared*, a new name is taken to be new to the schema and every
    other name to be declared.

    Returns the replacing payloads: one as a rule, none when the router routes
    nothing any more, several when it was split.

    Raises:
        ValueError: If a merge joins a class the router only looks up with one
            it writes; one class cannot be both.
    """
    mapping = {old: new for old, new in classes.items() if old != new}
    before = _router_view(payload)
    reach = router_reach(before, known_vertices=declared)
    known_before = (
        set(declared) if declared is not None else _assumed_declared(before, mapping)
    )
    _refuse_mixed_lookup(before, mapping, reach, known_before)
    if reach is not None:
        pieces = _projection_pieces(payload, mapping, reach, known_before)
        if len(pieces) > 1:
            closed = _closed_router(payload, reach)
            return [
                renamed
                for piece in pieces
                if not _routes_nothing(
                    renamed := _renamed_router(
                        _router_piece(closed, piece), mapping, reach
                    )
                )
            ]

    known_after = (known_before - set(mapping)) | {
        new for new in mapping.values() if new is not None
    }
    candidate = _renamed_router(payload, mapping, reach)
    if not _routes_as_before(
        before, candidate, mapping, reach, known_before, known_after
    ):
        if reach is None:
            logger.warning(
                "vertex_router on %r: renaming %s changes what it routes, and "
                "without the schema's classes it cannot be closed",
                before.get("type_field"),
                sorted(mapping),
            )
        else:
            candidate = _renamed_router(_closed_router(payload, reach), mapping, reach)
    return [] if _routes_nothing(candidate) else [candidate]

mark_find_in_pipeline(pipeline, vertex, selector)

pipeline with every production of vertex found by selector.

A vertex step for it gains find: <selector>; a vertex_router whose :func:router_reach includes it gains {vertex: selector} in its find map, its other classes untouched. lookup_only is left as is: a step that wrote the class now attaches to the node it finds, one that only referenced it now references it by selector. Steps keep their authored spelling, at every level.

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

def mark_find_in_pipeline(
    pipeline: list[dict[str, Any]], vertex: str, selector: str
) -> list[dict[str, Any]]:
    """*pipeline* with every production of *vertex* found by *selector*.

    A ``vertex`` step for it gains ``find: <selector>``; a ``vertex_router``
    whose :func:`router_reach` includes it gains ``{vertex: selector}`` in its
    ``find`` map, its other classes untouched. ``lookup_only`` is left as is:
    a step that wrote the class now attaches to the node it finds, one that
    only referenced it now references it by *selector*. Steps keep their
    authored spelling, at every level.
    """
    out = deepcopy(pipeline)
    _mark_find(out, vertex, selector)
    return out

mark_lookup_only_in_pipeline(pipeline, vertex)

pipeline with every production of vertex turned into a lookup.

A vertex step for it gains lookup_only: true. An open vertex_router routes an unmapped value as the class name, so it can produce vertex unless its vertex_types leaves it out, and a closed one (type_map_only) can when its table routes to it -- its :func:router_reach: such a router gains vertex in its lookup_only list, and the other classes it routes to are still written. Steps keep their authored spelling, at every level.

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

def mark_lookup_only_in_pipeline(
    pipeline: list[dict[str, Any]], vertex: str
) -> list[dict[str, Any]]:
    """*pipeline* with every production of *vertex* turned into a lookup.

    A ``vertex`` step for it gains ``lookup_only: true``. An open
    ``vertex_router`` routes an unmapped value as the class name, so it can
    produce *vertex* unless its ``vertex_types`` leaves it out, and a closed
    one (``type_map_only``) can when its table routes to it -- its
    :func:`router_reach`: such a router gains *vertex* in its ``lookup_only`` list,
    and the other classes it routes to are still written. Steps keep their
    authored spelling, at every level.
    """
    out = deepcopy(pipeline)
    _mark_lookup_only(out, vertex)
    return out

pass_through_classes(pipeline, vocabulary)

Classes of vocabulary the routers of pipeline reach only by pass-through.

Empty unless some router is open and unbounded. A class any router names -- a table target, or a vertex_types entry -- is excluded: the router already checked its from against that class at load.

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

def pass_through_classes(
    pipeline: list[dict[str, Any]], vocabulary: Collection[str]
) -> set[str]:
    """Classes of *vocabulary* the routers of *pipeline* reach only by pass-through.

    Empty unless some router is open and unbounded. A class any router names
    -- a table target, or a ``vertex_types`` entry -- is excluded: the router
    already checked its ``from`` against that class at load.
    """
    named: set[str] = set()
    unbounded = False

    def walk(step: Any) -> None:
        nonlocal unbounded
        if isinstance(step, list):
            for item in step:
                walk(item)
            return
        if not isinstance(step, dict):
            return
        normalized = normalize_actor_step(dict(step))
        if normalized.get("type") == "vertex_router":
            named.update((normalized.get("type_map") or {}).values())
            named.update(normalized.get("vertex_types") or ())
            if not normalized.get("type_map_only") and not normalized.get(
                "vertex_types"
            ):
                unbounded = True
        for container in (step.get("descend"), step):
            if isinstance(container, dict):
                for key in ("apply", "pipeline"):
                    walk(container.get(key))

    walk(pipeline)
    return set(vocabulary) - named if unbounded else set()

pipeline_mentions_any_vertex(steps, names)

Return True if any pipeline step references a vertex name in names.

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

def pipeline_mentions_any_vertex(steps: list[dict[str, Any]], names: set[str]) -> bool:
    """Return True if any pipeline step references a vertex name in *names*."""
    if not names:
        return False
    for step in steps:
        if not isinstance(step, dict):
            continue
        s = normalize_actor_step(dict(step))
        t = s.get("type")
        if t == "vertex":
            if s.get("vertex") in names:
                return True
        elif t == "vertex_router":
            tm = s.get("type_map") or {}
            if any(v in names for v in tm.values() if isinstance(v, str)):
                return True
            vfm = s.get("vertex_from_map") or {}
            if any(k in names for k in vfm if isinstance(k, str)):
                return True
            if any(v in names for v in s.get("vertex_types") or []):
                return True
        elif t == "edge":
            if any(_names_any(p, names) for p in _edge_payload_and_links(s)):
                return True
        elif t == "descend":
            pl = s.get("pipeline") or []
            if isinstance(pl, list) and pipeline_mentions_any_vertex(
                [cast_step(x) for x in pl if isinstance(x, dict)], names
            ):
                return True
    return False

rename_endpoint_selector_classes(payload, vertex_name)

Rename the classes of payload's per-class source_match / target_match.

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

def rename_endpoint_selector_classes(
    payload: dict[str, Any], vertex_name: Callable[[str], str]
) -> None:
    """Rename the classes of *payload*'s per-class ``source_match`` / ``target_match``."""
    for key in _MATCH_KEYS:
        if isinstance(payload.get(key), dict):
            payload[key] = _renamed_selector(payload[key], vertex_name)

renamed_lookup_classes(value, vertex_name)

A router's lookup_only list with its classes renamed (merges deduplicated).

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

def renamed_lookup_classes(value: Any, vertex_name: Callable[[str], str]) -> Any:
    """A router's ``lookup_only`` list with its classes renamed (merges deduplicated)."""
    if not isinstance(value, list):
        return value
    out: list[Any] = []
    for name in value:
        new = vertex_name(name) if isinstance(name, str) else name
        if new not in out:
            out.append(new)
    return out

rewrite_edge_endpoints_in_pipeline(pipeline, mapping)

Repoint edge steps whose EdgeId appears in mapping at new endpoints.

Keyed on the full (source, target, relation) triple rather than on vertex names, so an edge step between the same pair of types under a different relation is left alone.

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

def rewrite_edge_endpoints_in_pipeline(
    pipeline: list[dict[str, Any]],
    mapping: dict[tuple[str, str, str | None], tuple[str, str]],
) -> list[dict[str, Any]]:
    """Repoint edge steps whose ``EdgeId`` appears in *mapping* at new endpoints.

    Keyed on the full ``(source, target, relation)`` triple rather than on vertex
    names, so an edge step between the same pair of types under a different relation
    is left alone.
    """
    out = deepcopy(pipeline)
    if not mapping:
        return out
    _retarget_edges_in_step(out, mapping)
    return out

rewrite_edge_properties_in_pipeline(pipeline, *, renames_by_relation=None, removals_by_relation=None, renames_by_edge=None, removals_by_edge=None)

Rewrite edge actor properties declarations by relation or by triple.

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

def rewrite_edge_properties_in_pipeline(
    pipeline: list[dict[str, Any]],
    *,
    renames_by_relation: dict[str, dict[str, str]] | None = None,
    removals_by_relation: dict[str, set[str]] | None = None,
    renames_by_edge: Mapping[tuple[str, str, str | None], dict[str, str]] | None = None,
    removals_by_edge: Mapping[tuple[str, str, str | None], set[str]] | None = None,
) -> list[dict[str, Any]]:
    """Rewrite edge actor `properties` declarations by relation or by triple."""
    renames_ctx = renames_by_relation or {}
    removals_ctx = removals_by_relation or {}
    renames_edge_ctx = renames_by_edge or {}
    removals_edge_ctx = removals_by_edge or {}
    if not (renames_ctx or removals_ctx or renames_edge_ctx or removals_edge_ctx):
        return deepcopy(pipeline)

    def _rewrite_one(target: dict[str, Any], view: dict[str, Any]) -> None:
        relation = view.get("relation")
        if isinstance(relation, str):
            renames = dict(renames_ctx.get(relation, {}))
            removals = set(removals_ctx.get(relation, set()))
        else:
            renames = {}
            removals = set()
        for edge_id in _per_edge_reach(view, renames_edge_ctx):
            renames |= renames_edge_ctx[edge_id]
        for edge_id in _per_edge_reach(view, removals_edge_ctx):
            removals |= removals_edge_ctx[edge_id]
        _rewrite_edge_properties_payload(target, renames=renames, removals=removals)

    def _rewrite_edge_payload(payload: dict[str, Any]) -> None:
        _rewrite_one(payload, payload)
        links = payload.get("links")
        if isinstance(links, list):
            inherited = {
                key: payload[key] for key in _INHERITED_LINK_KEYS if key in payload
            }
            for link in links:
                if isinstance(link, dict):
                    _rewrite_one(link, {**inherited, **link})

    def _rewrite_step(step: dict[str, Any]) -> dict[str, Any]:
        out = deepcopy(step)
        for payload in edge_payloads(out):
            _rewrite_edge_payload(payload)
        descend_payload = out.get("descend")
        if isinstance(descend_payload, dict):
            nested_pipeline = descend_payload.get("pipeline")
            if isinstance(nested_pipeline, list):
                descend_payload["pipeline"] = [
                    _rewrite_step(item)
                    for item in nested_pipeline
                    if isinstance(item, dict)
                ]
        return out

    return [_rewrite_step(step) for step in pipeline if isinstance(step, dict)]

rewrite_endpoint_selectors_in_pipeline(pipeline, selectors)

Pin primary-identity edge endpoints of selectors keys to a secondary identity.

selectors maps a vertex name to the secondary identity name its endpoints should select. Used by ReplaceIdentityOp with endpoints: pin_to_retired so edge steps keep matching on the identity that was just retired, and by merge for a resource that references a re-keyed member. An endpoint a router role fills is pinned per class ({Class: selector}).

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

def rewrite_endpoint_selectors_in_pipeline(
    pipeline: list[dict[str, Any]], selectors: dict[str, str]
) -> list[dict[str, Any]]:
    """Pin primary-identity edge endpoints of *selectors* keys to a secondary identity.

    ``selectors`` maps a vertex name to the secondary identity name its endpoints
    should select. Used by ``ReplaceIdentityOp`` with ``endpoints: pin_to_retired`` so
    edge steps keep matching on the identity that was just retired, and by merge
    for a resource that references a re-keyed member. An endpoint a router role
    fills is pinned per class (``{Class: selector}``).
    """
    out = deepcopy(pipeline)
    if not selectors:
        return out
    _pin_endpoint_selectors_in_step(out, selectors, role_reach(out))
    return out

rewrite_entity_names_in_pipeline(step, *, vertices=None, edges=None, declared=None)

Mutate a pipeline payload in place to rename vertices/relations.

declared is the schema's classes before the rename; a router is evolved by :func:evolve_router against it.

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

def rewrite_entity_names_in_pipeline(
    step: Any,
    *,
    vertices: dict[str, str] | None = None,
    edges: dict[str, str] | None = None,
    declared: Collection[str] | None = None,
) -> None:
    """Mutate a pipeline payload in place to rename vertices/relations.

    *declared* is the schema's classes before the rename; a router is evolved
    by :func:`evolve_router` against it.
    """
    vertex_name = _build_name_transformer(vertices)
    edge_name = _build_name_transformer(edges)

    if isinstance(step, list):
        for item in step:
            rewrite_entity_names_in_pipeline(
                item,
                vertices=vertices,
                edges=edges,
                declared=declared,
            )
        return
    if not isinstance(step, dict):
        return

    if isinstance(step.get("vertex"), str):
        step["vertex"] = vertex_name(step["vertex"])

    payload = router_payload(step)
    if payload is not None and vertices:
        evolved = evolve_router(payload, vertices, declared=declared)
        if len(evolved) > 1:
            raise ValueError(
                f"vertex_router on {payload.get('type_field')!r}: renaming "
                f"{sorted(vertices)} folds classes it projects differently; "
                "merge them with merge_vertices or canonicalize instead"
            )
        if evolved:
            payload.clear()
            payload.update(evolved[0])

    edge_payload = step.get("edge")
    if isinstance(edge_payload, dict):
        _rewrite_entity_names_in_edge_step(
            edge_payload,
            vertex_name=vertex_name,
            edge_name=edge_name,
        )
    elif step.get("type") == "edge" or _is_untyped_flat_edge(step):
        # Flat form: the edge payload *is* the step. Only string-valued endpoint keys
        # are touched, so a vertex step's dict-valued ``from`` column map is unaffected.
        # The untyped spelling (``{from, to, relation}``) is an edge too: it is how
        # ``normalize_actor_step`` reads it, so a rename that skipped it would leave
        # the ingested edge pointing at a vertex that no longer exists.
        _rewrite_entity_names_in_edge_step(
            step,
            vertex_name=vertex_name,
            edge_name=edge_name,
        )

    create_edge_payload = step.get("create_edge")
    if isinstance(create_edge_payload, dict):
        _rewrite_entity_names_in_edge_step(
            create_edge_payload,
            vertex_name=vertex_name,
            edge_name=edge_name,
        )

    descend_payload = step.get("descend")
    if isinstance(descend_payload, dict):
        apply_payload = descend_payload.get("apply")
        if apply_payload is not None:
            rewrite_entity_names_in_pipeline(
                apply_payload,
                vertices=vertices,
                edges=edges,
                declared=declared,
            )
        pipeline_payload = descend_payload.get("pipeline")
        if pipeline_payload is not None:
            rewrite_entity_names_in_pipeline(
                pipeline_payload,
                vertices=vertices,
                edges=edges,
                declared=declared,
            )

    if isinstance(step.get("apply"), list):
        rewrite_entity_names_in_pipeline(
            step["apply"],
            vertices=vertices,
            edges=edges,
            declared=declared,
        )
    if isinstance(step.get("pipeline"), list):
        rewrite_entity_names_in_pipeline(
            step["pipeline"],
            vertices=vertices,
            edges=edges,
            declared=declared,
        )

rewrite_extra_weights_vertex_field_names(entries, renames_by_vertex)

Rewrite extra_weights[*].vertex_weights for vertex field renames.

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

def rewrite_extra_weights_vertex_field_names(
    entries: list[Any],
    renames_by_vertex: dict[str, dict[str, str]],
) -> list[dict[str, Any]]:
    """Rewrite ``extra_weights[*].vertex_weights`` for vertex field renames."""
    if not entries:
        return []
    result: list[dict[str, Any]] = []
    for entry in entries:
        d = (
            dict(entry)
            if isinstance(entry, dict)
            else entry.to_dict(skip_defaults=False)
        )
        vw = d.get("vertex_weights")
        if isinstance(vw, list) and renames_by_vertex:
            d["vertex_weights"] = rewrite_vertex_weights_vertex_field_names(
                vw, renames_by_vertex
            )
        result.append(d)
    return result

rewrite_remove_edge_ids_in_pipeline(pipeline, removed_edge_ids)

Drop edge/create_edge steps (and links) targeting removed edge triples.

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

def rewrite_remove_edge_ids_in_pipeline(
    pipeline: list[dict[str, Any]],
    removed_edge_ids: set[tuple[str, str, str | None]],
) -> list[dict[str, Any]]:
    """Drop edge/create_edge steps (and links) targeting removed edge triples."""
    if not removed_edge_ids:
        return deepcopy(pipeline)

    def _removed(payload: dict[str, Any]) -> bool:
        return _payload_targets_removed_edge(payload, removed_edge_ids)

    def _prune(payload: dict[str, Any]) -> None:
        _prune_relation_map_for_removed_edge_ids(payload, removed_edge_ids)
        links = payload.get("links")
        if isinstance(links, list):
            payload["links"] = [
                link
                for link in links
                if not (isinstance(link, dict) and _removed(link))
            ]

    return _drop_edge_payloads(pipeline, _removed, _prune)

rewrite_remove_relations_in_pipeline(pipeline, removed_relations)

Drop edge/create_edge steps (and links) targeting removed relations.

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

def rewrite_remove_relations_in_pipeline(
    pipeline: list[dict[str, Any]], removed_relations: set[str]
) -> list[dict[str, Any]]:
    """Drop edge/create_edge steps (and links) targeting removed relations."""
    if not removed_relations:
        return deepcopy(pipeline)

    def _removed(payload: dict[str, Any]) -> bool:
        return payload.get("relation") in removed_relations

    def _prune(payload: dict[str, Any]) -> None:
        relation_map = payload.get("relation_map")
        if isinstance(relation_map, dict):
            payload["relation_map"] = {
                k: v
                for k, v in relation_map.items()
                if not (isinstance(v, str) and v in removed_relations)
            }
        links = payload.get("links")
        if isinstance(links, list):
            payload["links"] = [
                link
                for link in links
                if not (isinstance(link, dict) and _removed(link))
            ]

    return _drop_edge_payloads(pipeline, _removed, _prune)

rewrite_remove_vertex_properties_in_pipeline(pipeline, removals, *, declared=None)

Remove references to dropped vertex fields from pipeline steps.

A router stops mapping them for each class it reaches -- declared is the schema's classes, the ones an unbounded router reaches -- in that class's own vertex_from_map entry, so the classes sharing its from keep it.

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

def rewrite_remove_vertex_properties_in_pipeline(
    pipeline: list[dict[str, Any]],
    removals: dict[str, set[str]],
    *,
    declared: Collection[str] | None = None,
) -> list[dict[str, Any]]:
    """Remove references to dropped vertex fields from pipeline steps.

    A router stops mapping them for each class it reaches -- *declared* is the
    schema's classes, the ones an unbounded router reaches -- in that class's
    own ``vertex_from_map`` entry, so the classes sharing its ``from`` keep it.
    """
    if not removals:
        return deepcopy(pipeline)

    def _without(
        removed: set[str],
    ) -> Callable[[dict[str, Any] | None], dict[str, Any] | None]:
        def drop(current: dict[str, Any] | None) -> dict[str, Any] | None:
            if current is None or not set(current) & removed:
                return None
            return {k: v for k, v in current.items() if k not in removed}

        return drop

    def _rewrite_step(step: dict[str, Any]) -> dict[str, Any]:
        out = deepcopy(normalize_actor_step(dict(step)))
        step_type = out.get("type")

        if step_type == "vertex_router":
            return _rewrite_router_projections(
                step,
                {v: _without(fields) for v, fields in removals.items() if fields},
                declared,
            )

        if step_type == "vertex":
            vertex_name = out.get("vertex")
            if isinstance(vertex_name, str):
                removed = removals.get(vertex_name, set())
                if removed:
                    from_map = out.get("from")
                    if isinstance(from_map, dict):
                        out["from"] = {
                            key: value
                            for key, value in from_map.items()
                            if isinstance(key, str) and key not in removed
                        }
                    keep_fields = out.get("keep_fields")
                    if isinstance(keep_fields, list):
                        out["keep_fields"] = [
                            key
                            for key in keep_fields
                            if not (isinstance(key, str) and key in removed)
                        ]

        elif step_type == "transform":
            rename_map = out.get("rename")
            if isinstance(rename_map, dict):
                blocked_fields = set().union(*removals.values())
                out["rename"] = {
                    key: value
                    for key, value in rename_map.items()
                    if not (isinstance(value, str) and value in blocked_fields)
                }

        elif step_type == "edge":
            weights = out.get("vertex_weights")
            if isinstance(weights, list):
                filtered_weights: list[dict[str, Any]] = []
                for entry in weights:
                    if not isinstance(entry, dict):
                        continue
                    name = entry.get("name")
                    if not isinstance(name, str):
                        filtered_weights.append(dict(entry))
                        continue
                    removed = removals.get(name, set())
                    if not removed:
                        filtered_weights.append(dict(entry))
                        continue
                    rewritten = dict(entry)
                    fields = rewritten.get("fields")
                    if isinstance(fields, list):
                        rewritten["fields"] = [
                            f
                            for f in fields
                            if not (isinstance(f, str) and f in removed)
                        ]
                    map_payload = rewritten.get("map")
                    if isinstance(map_payload, dict):
                        rewritten["map"] = {
                            k: v
                            for k, v in map_payload.items()
                            if not (isinstance(k, str) and k in removed)
                        }
                    filter_payload = rewritten.get("filter")
                    if isinstance(filter_payload, dict):
                        rewritten["filter"] = {
                            k: v
                            for k, v in filter_payload.items()
                            if not (isinstance(k, str) and k in removed)
                        }
                    filtered_weights.append(rewritten)
                out["vertex_weights"] = filtered_weights

        elif step_type == "descend":
            nested = out.get("pipeline")
            if isinstance(nested, list):
                out["pipeline"] = [
                    _rewrite_step(item) for item in nested if isinstance(item, dict)
                ]

        return out

    return [_rewrite_step(step) for step in pipeline if isinstance(step, dict)]

rewrite_remove_vertices_in_pipeline(pipeline, removed, *, declared=None)

Trim pipeline to what survives removing the classes in removed.

Step-wise, never resource-wise. A vertex or edge step naming a removed class goes. A vertex_router is evolved by :func:evolve_router against declared (the schema's classes before the removal): it loses its entries for the class and keeps routing the rest, and a router left routing nothing goes. The class also leaves the per-class selectors of the edge steps that survive. An edge step addressing a role that no step fills any more goes too. A descend stays while anything survives under it; transforms are left as authored. Untouched steps keep their authored spelling; a trimmed router is edited in place, and a descend that lost a step comes back normalized.

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

def rewrite_remove_vertices_in_pipeline(
    pipeline: list[dict[str, Any]],
    removed: set[str],
    *,
    declared: Collection[str] | None = None,
) -> list[dict[str, Any]]:
    """Trim *pipeline* to what survives removing the classes in *removed*.

    Step-wise, never resource-wise. A ``vertex`` or ``edge`` step naming a
    removed class goes. A ``vertex_router`` is evolved by
    :func:`evolve_router` against *declared* (the schema's classes before the
    removal): it loses its entries for the class and keeps routing the rest,
    and a router left routing nothing goes. The class also leaves the
    per-class selectors of the edge steps that survive. An edge step
    addressing a role that no step fills any more goes too. A ``descend``
    stays while anything survives under it; transforms are left as authored.
    Untouched steps keep their authored spelling; a trimmed router is edited
    in place, and a ``descend`` that lost a step comes back normalized.
    """
    if not removed:
        return deepcopy(pipeline)
    removal: dict[str, str | None] = dict.fromkeys(removed)

    def _rewrite_step(step: dict[str, Any]) -> dict[str, Any] | None:
        normalized = normalize_actor_step(dict(step))
        step_type = normalized.get("type")
        if step_type == "vertex":
            return None if normalized.get("vertex") in removed else deepcopy(step)
        if step_type == "edge":
            if _names_any(normalized, removed):
                return None
            out = deepcopy(step)
            for edge_payload in edge_payloads(out):
                if not _keep_links(
                    edge_payload, lambda link: _names_any(link, removed)
                ):
                    return None
                _drop_selector_classes(edge_payload, removed)
            return out
        if step_type == "vertex_router":
            out = deepcopy(step)
            payload = router_payload(out)
            if payload is None:
                return out
            evolved = evolve_router(payload, removal, declared=declared)
            if not evolved:
                return None
            # A removal folds no classes, so it never splits a router.
            payload.clear()
            payload.update(evolved[0])
            return out
        if step_type == "descend":
            nested = normalized.get("pipeline")
            if not isinstance(nested, list):
                return deepcopy(step)
            items = [item for item in nested if isinstance(item, dict)]
            survivors = [
                rewritten
                for rewritten in (_rewrite_step(item) for item in items)
                if rewritten is not None
            ]
            if not survivors:
                return None
            if survivors == items:
                return deepcopy(step)
            out = deepcopy(normalized)
            out["pipeline"] = survivors
            return out
        return deepcopy(step)

    out = [
        rewritten
        for rewritten in (
            _rewrite_step(step) for step in pipeline if isinstance(step, dict)
        )
        if rewritten is not None
    ]
    unfilled = set(role_reach(pipeline)) - {
        role for role, held in role_reach(out).items() if held != frozenset()
    }
    return _drop_edges_on_roles(out, unfilled) if unfilled else out

rewrite_vertex_field_names_in_pipeline(pipeline, renames, *, available_vertices=None, declared=None)

Rewrite vertex field names across a resource pipeline.

Walks the dict pipeline (no runtime tree mutation):

• vertex steps: ensure from: covers the rename. Existing {old_field: doc_col} becomes {new_field: doc_col}; missing entries are injected as {new_field: old_field} so the doc can still address the property by its original name.
• transform steps with rename: rewrite values that pointed at renamed vertex fields. call steps are unchanged.
• edge steps: rewrite vertex_weights field/map/filter keys per Weight.name.
• vertex_router steps: the same from rewrite, per class the router reaches, in that class's vertex_from_map entry.
• descend steps: recurse with an extended available_vertices set.

available_vertices is the set of vertex names visible from the parent scope. Vertex names introduced at the current level are added automatically. declared is the schema's classes: the ones an unbounded router reaches.

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

def rewrite_vertex_field_names_in_pipeline(
    pipeline: list[dict[str, Any]],
    renames: dict[str, dict[str, str]],
    *,
    available_vertices: set[str] | None = None,
    declared: Collection[str] | None = None,
) -> list[dict[str, Any]]:
    """Rewrite vertex field names across a resource pipeline.

    Walks the dict pipeline (no runtime tree mutation):

    - ``vertex`` steps: ensure ``from:`` covers the rename. Existing
      ``{old_field: doc_col}`` becomes ``{new_field: doc_col}``; missing entries
      are injected as ``{new_field: old_field}`` so the doc can still address
      the property by its original name.
    - ``transform`` steps with ``rename``: rewrite values that pointed at renamed
      vertex fields. ``call`` steps are unchanged.
    - ``edge`` steps: rewrite ``vertex_weights`` field/map/filter keys per
      ``Weight.name``.
    - ``vertex_router`` steps: the same ``from`` rewrite, per class the router
      reaches, in that class's ``vertex_from_map`` entry.
    - ``descend`` steps: recurse with an extended ``available_vertices`` set.

    ``available_vertices`` is the set of vertex names visible from the parent
    scope. Vertex names introduced at the current level are added automatically.
    *declared* is the schema's classes: the ones an unbounded router reaches.
    """
    if not renames:
        return deepcopy(pipeline)
    parent_scope = set(available_vertices) if available_vertices else set()
    level_vertices = _collect_level_vertices(pipeline, declared=declared)
    scope = parent_scope | level_vertices
    return [
        _rewrite_vertex_field_step(step, renames, scope, declared)
        for step in pipeline
        if isinstance(step, dict)
    ]

rewrite_vertex_names_in_pipeline(pipeline, mapping, *, declared=None)

Rewrite all steps in a resource pipeline (see :func:rewrite_vertex_names_in_step).

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

def rewrite_vertex_names_in_pipeline(
    pipeline: list[dict[str, Any]],
    mapping: dict[str, str],
    *,
    declared: Collection[str] | None = None,
) -> list[dict[str, Any]]:
    """Rewrite all steps in a resource pipeline (see :func:`rewrite_vertex_names_in_step`)."""
    if not mapping:
        return deepcopy(pipeline)
    return [
        rewritten
        for s in pipeline
        for rewritten in _rewrite_vertex_names(s, mapping, declared=declared)
    ]

rewrite_vertex_names_in_step(step, mapping, *, declared=None)

Return a deep-copied step with vertex names rewritten per mapping.

declared is the schema's classes before the rewrite; a router is evolved by :func:evolve_router against it.

Raises:

Type	Description
ValueError	If the rewrite splits a router into several steps; use :func:rewrite_vertex_names_in_pipeline, which can hold them.

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

def rewrite_vertex_names_in_step(
    step: dict[str, Any],
    mapping: dict[str, str],
    *,
    declared: Collection[str] | None = None,
) -> dict[str, Any]:
    """Return a deep-copied step with vertex names rewritten per *mapping*.

    *declared* is the schema's classes before the rewrite; a router is evolved
    by :func:`evolve_router` against it.

    Raises:
        ValueError: If the rewrite splits a router into several steps; use
            :func:`rewrite_vertex_names_in_pipeline`, which can hold them.
    """
    steps = _rewrite_vertex_names(step, mapping, declared=declared)
    if len(steps) != 1:
        raise ValueError(
            f"rewriting {sorted(mapping)} turns one step into {len(steps)}"
        )
    return steps[0]

rewrite_vertex_names_in_value(obj, mapping)

Deep-rewrite obj (pipelines, infer specs, extra_weights, nested dicts).

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

def rewrite_vertex_names_in_value(obj: Any, mapping: dict[str, str]) -> Any:
    """Deep-rewrite *obj* (pipelines, infer specs, extra_weights, nested dicts)."""
    if not mapping:
        return deepcopy(obj) if isinstance(obj, (dict, list)) else obj
    if isinstance(obj, list):
        return [rewrite_vertex_names_in_value(x, mapping) for x in obj]
    if isinstance(obj, dict):
        if "edge" in obj and isinstance(obj["edge"], dict):
            inner = deepcopy(obj)
            inner["edge"] = rewrite_vertex_names_in_value(obj["edge"], mapping)
            # An extra_weights entry carries vertex_weights alongside its edge.
            rewrite_vertex_weight_names(inner, lambda name: mapping.get(name, name))
            return inner
        t = obj.get("type")
        if t in ("vertex", "edge", "descend", "vertex_router"):
            return rewrite_vertex_names_in_step(obj, mapping)
        if t == "transform":
            return deepcopy(obj)
        if all(k in obj for k in ("source", "target")):
            out = deepcopy(obj)
            src = out.get("source")
            tgt = out.get("target")
            if isinstance(src, str) and src in mapping:
                out["source"] = mapping[src]
            if isinstance(tgt, str) and tgt in mapping:
                out["target"] = mapping[tgt]
            rewrite_vertex_weight_names(out, lambda name: mapping.get(name, name))
            rename_endpoint_selector_classes(out, lambda name: mapping.get(name, name))
            return out
        if "vertex" in obj and isinstance(obj["vertex"], str) and t is None:
            out = deepcopy(obj)
            if out["vertex"] in mapping:
                out["vertex"] = mapping[out["vertex"]]
            return out
        return {k: rewrite_vertex_names_in_value(v, mapping) for k, v in obj.items()}
    return obj

rewrite_vertex_weight_names(payload, vertex_name)

Rewrite vertex_weights[].name in place.

Weight.name on an edge step is a vertex name — it selects which endpoint's observation columns the weight reads — and collect_vertex_names_from_pipeline counts it as a vertex reference. Missing it leaves a pipeline pointing at a type the schema no longer has.

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

def rewrite_vertex_weight_names(
    payload: dict[str, Any], vertex_name: Callable[[str], str]
) -> None:
    """Rewrite ``vertex_weights[].name`` in place.

    ``Weight.name`` on an edge step is a *vertex* name — it selects which endpoint's
    observation columns the weight reads — and ``collect_vertex_names_from_pipeline``
    counts it as a vertex reference. Missing it leaves a pipeline pointing at a type
    the schema no longer has.
    """
    weights = payload.get("vertex_weights")
    if not isinstance(weights, list):
        return
    for weight in weights:
        if isinstance(weight, dict) and isinstance(weight.get("name"), str):
            weight["name"] = vertex_name(weight["name"])

rewrite_vertex_weights_vertex_field_names(weights, renames_by_vertex)

Rewrite :class:~graflo.architecture.graph_types.Weight references to vertex fields.

Each weight's name selects the logical vertex whose renames_by_vertex[name] map applies (old field name -> new field name). filter and map keys name vertex fields and follow the rename.

The edge attribute a weight writes does not follow it: that name belongs to the edge, and renaming it is rename_edge_properties' job. A renamed entry of fields -- whose attribute name is the vertex field's name -- is therefore moved to map as {new_field: old_attribute} -- unless map already uses that key, in which case it cannot name a second attribute and the fields entry is renamed instead (with a warning).

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

def rewrite_vertex_weights_vertex_field_names(
    weights: list[Any],
    renames_by_vertex: dict[str, dict[str, str]],
) -> list[dict[str, Any]]:
    """Rewrite :class:`~graflo.architecture.graph_types.Weight` references to vertex fields.

    Each weight's ``name`` selects the logical vertex whose ``renames_by_vertex[name]``
    map applies (old field name -> new field name). ``filter`` and ``map`` keys
    name vertex fields and follow the rename.

    The edge attribute a weight writes does **not** follow it: that name belongs
    to the edge, and renaming it is ``rename_edge_properties``' job. A renamed
    entry of ``fields`` -- whose attribute name *is* the vertex field's name -- is
    therefore moved to ``map`` as ``{new_field: old_attribute}`` -- unless ``map``
    already uses that key, in which case it cannot name a second attribute and the
    ``fields`` entry is renamed instead (with a warning).
    """
    from graflo.architecture.graph_types import Weight

    if not weights:
        return []
    out: list[dict[str, Any]] = []
    for raw in weights:
        w = Weight.model_validate(raw)
        per = {}
        vn = w.name
        if isinstance(vn, str) and vn in renames_by_vertex:
            per = renames_by_vertex[vn]
        if per:

            def _remap_obs_key(obs_key: Any, _per: dict = per) -> Any:
                if isinstance(obs_key, str):
                    return _per.get(obs_key, obs_key)
                return obs_key

            new_map = {_remap_obs_key(k): v for k, v in dict(w.map).items()}
            new_fields: list[Any] = []
            for fname in w.fields:
                if not isinstance(fname, str) or fname not in per:
                    new_fields.append(fname)
                elif per[fname] not in new_map:
                    new_map[per[fname]] = w.cfield(fname)
                else:
                    # The field also feeds a `map` entry, and one map key cannot
                    # name two attributes: the fields-derived one follows the rename.
                    logger.warning(
                        "vertex_weights on %r: %r is both in `fields` and a `map` "
                        "key, so its `fields` attribute is renamed to %r",
                        vn,
                        fname,
                        w.cfield(per[fname]),
                    )
                    new_fields.append(per[fname])
            new_filter = {_remap_obs_key(k): v for k, v in dict(w.filter).items()}
            w = w.model_copy(
                update={"fields": new_fields, "map": new_map, "filter": new_filter}
            )
        out.append(w.to_dict(skip_defaults=False))
    return out

router_payload(step)

The dict holding step's router fields, or None for another step.

Nested under vertex_router in the shorthand spelling, the step itself in the flat one; editing it keeps the authored spelling.

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

def router_payload(step: dict[str, Any]) -> dict[str, Any] | None:
    """The dict holding *step*'s router fields, or ``None`` for another step.

    Nested under ``vertex_router`` in the shorthand spelling, the step itself
    in the flat one; editing it keeps the authored spelling.
    """
    nested = step.get("vertex_router")
    if isinstance(nested, dict):
        return nested
    if step.get("type") == "vertex_router" or (
        "type_field" in step and "type" not in step
    ):
        return step
    return None