graflo.architecture.evolution.db_profile

Update :class:~graflo.architecture.schema.database_features.DatabaseProfile after vertex changes.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

Functions:

apply_edge_id_removal_to_db_profile(profile, removed_edge_ids)

Drop edge specs/default values for removed logical edge triples.

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

def apply_edge_id_removal_to_db_profile(
    profile: DatabaseProfile, removed_edge_ids: set[EdgeId]
) -> None:
    """Drop edge specs/default values for removed logical edge triples."""
    if not removed_edge_ids:
        return
    profile.edge_specs = [
        spec for spec in profile.edge_specs if spec.edge_id not in removed_edge_ids
    ]
    dpv = profile.default_property_values
    if dpv is None:
        return
    object.__setattr__(
        dpv,
        "edges",
        [edge for edge in dpv.edges if edge.edge_id not in removed_edge_ids],
    )

apply_edge_property_removal_to_db_profile(profile, removals_by_relation, *, removals_by_edge=None)

Remove edge property references from edge indexes/default values.

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

def apply_edge_property_removal_to_db_profile(
    profile: DatabaseProfile,
    removals_by_relation: dict[str, set[str]],
    *,
    removals_by_edge: Mapping[EdgeId, set[str]] | None = None,
) -> None:
    """Remove edge property references from edge indexes/default values."""
    by_edge = removals_by_edge or {}
    if not removals_by_relation and not by_edge:
        return

    def _removals(ref: EdgeRef) -> set[str]:
        out = (
            set(removals_by_relation.get(ref.relation, set()))
            if ref.relation is not None
            else set()
        )
        return out | by_edge.get(ref.edge_id, set())

    updated_specs: list[EdgePhysicalSpec] = []
    for spec in profile.edge_specs:
        removals = _removals(spec)
        if not removals:
            updated_specs.append(spec)
            continue
        indexes: list[Index] = []
        for index in spec.indexes:
            fields = [field for field in index.fields if field not in removals]
            if fields:
                indexes.append(index.model_copy(update={"fields": fields}, deep=True))
        updated_specs.append(spec.model_copy(update={"indexes": indexes}, deep=True))
    profile.edge_specs = updated_specs

    dpv = profile.default_property_values
    if dpv is None:
        return
    new_edges: list[EdgePropertyDefaults] = []
    for edge in dpv.edges:
        removals = _removals(edge)
        if not removals:
            new_edges.append(edge)
            continue
        values = {
            name: value for name, value in edge.values.items() if name not in removals
        }
        new_edges.append(edge.model_copy(update={"values": values}, deep=True))
    object.__setattr__(dpv, "edges", new_edges)

apply_edge_property_rename_to_db_profile(profile, renames_by_relation, *, renames_by_edge=None)

Rename edge property references in edge indexes/default values.

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

def apply_edge_property_rename_to_db_profile(
    profile: DatabaseProfile,
    renames_by_relation: dict[str, dict[str, str]],
    *,
    renames_by_edge: Mapping[EdgeId, dict[str, str]] | None = None,
) -> None:
    """Rename edge property references in edge indexes/default values."""
    by_edge = renames_by_edge or {}
    if not renames_by_relation and not by_edge:
        return

    def _renames(ref: EdgeRef) -> dict[str, str]:
        out = (
            dict(renames_by_relation.get(ref.relation, {}))
            if ref.relation is not None
            else {}
        )
        return out | by_edge.get(ref.edge_id, {})

    new_specs: list[EdgePhysicalSpec] = []
    for spec in profile.edge_specs:
        renames = _renames(spec)
        new_specs.append(
            spec.model_copy(
                update={
                    "indexes": _rewrite_index_fields(list(spec.indexes), renames),
                    "property_names": {
                        renames.get(logical, logical): physical
                        for logical, physical in spec.property_names.items()
                    },
                },
                deep=True,
            )
        )
    profile.edge_specs = new_specs
    dpv = profile.default_property_values
    if dpv is None:
        return
    new_edges: list[EdgePropertyDefaults] = []
    for edge in dpv.edges:
        renames = _renames(edge)
        values = {renames.get(name, name): value for name, value in edge.values.items()}
        new_edges.append(edge.model_copy(update={"values": values}, deep=True))
    object.__setattr__(dpv, "edges", new_edges)

apply_field_rename_to_db_profile(profile, renames)

Rewrite vertex field names referenced by a :class:DatabaseProfile.

renames maps each vertex name to a per-vertex {old_field: new_field} map. Updates:

• profile.vertex_indexes[vertex_name] field tuples.
• profile.vertex_property_names[vertex_name] keys.
• profile.default_property_values.vertices[vertex_name] keys.

Edge specs are left alone: their index fields name edge attributes, and renaming a vertex property does not rename an edge attribute -- not even one copied from that property by vertex_weights, which keeps its name.

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

def apply_field_rename_to_db_profile(
    profile: DatabaseProfile,
    renames: dict[str, dict[str, str]],
) -> None:
    """Rewrite vertex field names referenced by a :class:`DatabaseProfile`.

    ``renames`` maps each vertex name to a per-vertex ``{old_field: new_field}``
    map. Updates:

    - ``profile.vertex_indexes[vertex_name]`` field tuples.
    - ``profile.vertex_property_names[vertex_name]`` keys.
    - ``profile.default_property_values.vertices[vertex_name]`` keys.

    Edge specs are left alone: their index fields name *edge* attributes, and
    renaming a vertex property does not rename an edge attribute -- not even
    one copied from that property by ``vertex_weights``, which keeps its name.
    """
    if not renames:
        return

    new_vertex_indexes: dict[str, list[Index]] = {}
    for vertex_name, indexes in profile.vertex_indexes.items():
        per_vertex = renames.get(vertex_name) or {}
        new_vertex_indexes[vertex_name] = _rewrite_index_fields(
            list(indexes), per_vertex
        )
    profile.vertex_indexes = new_vertex_indexes

    # A physical name belongs to the property, not to its logical name.
    new_vpn: dict[str, dict[str, str]] = {}
    for vertex_name, names in profile.vertex_property_names.items():
        per_vertex = renames.get(vertex_name) or {}
        new_vpn[vertex_name] = {
            per_vertex.get(logical, logical): physical
            for logical, physical in names.items()
        }
    profile.vertex_property_names = new_vpn

    dpv = profile.default_property_values
    if dpv is None:
        return

    new_dpv_vertices: dict[str, dict[str, Any]] = {}
    for vertex_name, props in dpv.vertices.items():
        per_vertex = renames.get(vertex_name) or {}
        new_props: dict[str, Any] = {}
        for prop_name, value in props.items():
            new_props[per_vertex.get(prop_name, prop_name)] = value
        new_dpv_vertices[vertex_name] = new_props
    object.__setattr__(dpv, "vertices", new_dpv_vertices)

apply_inverse_edges_to_db_profile(profile, relation_map, created)

Give each created inverse edge a copy of its forward edge's specs and defaults.

Only edges the op created are touched: an inverse edge that already existed keeps whatever physical profile its author gave it. relation_name is not copied -- it is the storage type name, and copying it would store the forward and inverse relations as one physical type. Native inverses are keyed by relation and are refused alongside explicit inverse edges, so there is none to copy.

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

def apply_inverse_edges_to_db_profile(
    profile: DatabaseProfile,
    relation_map: dict[str, str],
    created: set[EdgeId],
) -> None:
    """Give each created inverse edge a copy of its forward edge's specs and defaults.

    Only edges the op created are touched: an inverse edge that already existed
    keeps whatever physical profile its author gave it. ``relation_name`` is not
    copied -- it is the storage type name, and copying it would store the
    forward and inverse relations as one physical type. Native inverses are keyed by relation
    and are refused alongside explicit inverse edges, so there is none to copy.
    """
    if not relation_map or not created:
        return

    def _created_inverse(
        source: str, target: str, relation: str | None
    ) -> EdgeId | None:
        if relation is None or relation not in relation_map:
            return None
        inverse_id: EdgeId = (target, source, relation_map[relation])
        return inverse_id if inverse_id in created else None

    existing_spec_keys = {spec.physical_key for spec in profile.edge_specs}
    new_specs = list(profile.edge_specs)
    for spec in profile.edge_specs:
        inverse_id = _created_inverse(spec.source, spec.target, spec.relation)
        if inverse_id is None:
            continue
        inverse_spec = EdgePhysicalSpec(
            source=inverse_id[0],
            target=inverse_id[1],
            relation=inverse_id[2],
            purpose=spec.purpose,
            indexes=[idx.model_copy(deep=True) for idx in spec.indexes],
            indexes_mode=spec.indexes_mode,
        )
        if inverse_spec.physical_key not in existing_spec_keys:
            new_specs.append(inverse_spec)
            existing_spec_keys.add(inverse_spec.physical_key)
    profile.edge_specs = new_specs

    dpv = profile.default_property_values
    if dpv is None:
        return
    existing_default_ids = {entry.edge_id for entry in dpv.edges}
    new_defaults = list(dpv.edges)
    for entry in dpv.edges:
        inverse_id = _created_inverse(entry.source, entry.target, entry.relation)
        if inverse_id is None or inverse_id in existing_default_ids:
            continue
        new_defaults.append(
            EdgePropertyDefaults(
                source=inverse_id[0],
                target=inverse_id[1],
                relation=inverse_id[2],
                values=dict(entry.values),
            )
        )
        existing_default_ids.add(inverse_id)
    object.__setattr__(dpv, "edges", new_defaults)

apply_relation_removal_to_db_profile(profile, removed_relations)

Drop edge specs/default values for removed relation names.

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

def apply_relation_removal_to_db_profile(
    profile: DatabaseProfile, removed_relations: set[str]
) -> None:
    """Drop edge specs/default values for removed relation names."""
    if not removed_relations:
        return
    profile.edge_specs = [
        spec for spec in profile.edge_specs if spec.relation not in removed_relations
    ]
    retain_native_inverses(profile, set(profile.native_inverses) - removed_relations)
    dpv = profile.default_property_values
    if dpv is None:
        return
    object.__setattr__(
        dpv,
        "edges",
        [edge for edge in dpv.edges if edge.relation not in removed_relations],
    )

apply_relation_rename_to_db_profile(profile, relation_renames)

Rename logical edge relation keys in edge specs/default edge values.

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

def apply_relation_rename_to_db_profile(
    profile: DatabaseProfile, relation_renames: dict[str, str]
) -> None:
    """Rename logical edge relation keys in edge specs/default edge values."""
    if not relation_renames:
        return
    new_specs: list[EdgePhysicalSpec] = []
    for spec in profile.edge_specs:
        new_relation = (
            relation_renames.get(spec.relation, spec.relation)
            if spec.relation is not None
            else spec.relation
        )
        new_specs.append(
            spec.model_copy(
                update={"relation": new_relation},
                deep=True,
            )
        )
    profile.edge_specs = new_specs
    profile.native_inverses = [
        relation_renames.get(relation, relation) for relation in profile.native_inverses
    ]

    dpv = profile.default_property_values
    if dpv is None:
        return
    object.__setattr__(
        dpv,
        "edges",
        [
            edge.model_copy(
                update={
                    "relation": (
                        relation_renames.get(edge.relation, edge.relation)
                        if edge.relation is not None
                        else edge.relation
                    )
                },
                deep=True,
            )
            for edge in dpv.edges
        ],
    )

apply_vertex_merge_to_db_profile(profile, from_vertices, into)

Remap logical vertex keys in profile when merging from_vertices into into.

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

def apply_vertex_merge_to_db_profile(
    profile: DatabaseProfile,
    from_vertices: set[str],
    into: str,
) -> None:
    """Remap logical vertex keys in *profile* when merging *from_vertices* into *into*."""
    remap_vertices_in_db_profile(profile, {v: into for v in from_vertices})

apply_vertex_removal_to_db_profile(profile, removed)

Drop logical vertex keys and edge entries that reference removed vertices.

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

def apply_vertex_removal_to_db_profile(
    profile: DatabaseProfile, removed: set[str]
) -> None:
    """Drop logical vertex keys and edge entries that reference removed vertices."""
    if not removed:
        return
    for name in removed:
        profile.vertex_storage_names.pop(name, None)
        profile.vertex_property_names.pop(name, None)
        profile.vertex_indexes.pop(name, None)

    dpv = profile.default_property_values
    if dpv is not None:
        for name in removed:
            dpv.vertices.pop(name, None)
        dpv.edges = [
            e for e in dpv.edges if e.source not in removed and e.target not in removed
        ]

    profile.edge_specs = [
        s
        for s in profile.edge_specs
        if s.source not in removed and s.target not in removed
    ]

apply_vertex_rename_to_db_profile(profile, vertex_renames)

Rename logical vertex keys in profile.

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

def apply_vertex_rename_to_db_profile(
    profile: DatabaseProfile, vertex_renames: dict[str, str]
) -> None:
    """Rename logical vertex keys in *profile*."""
    if not vertex_renames:
        return

    new_vs: dict[str, str] = {}
    for k, v in profile.vertex_storage_names.items():
        nk = vertex_renames.get(k, k)
        if nk in new_vs and new_vs[nk] != v:
            raise ValueError(
                f"Conflicting vertex_storage_names for logical {nk!r}: "
                f"{new_vs[nk]!r} vs {v!r}"
            )
        new_vs[nk] = v
    profile.vertex_storage_names = new_vs

    new_vi: dict[str, list[Any]] = {}
    for k, vlist in profile.vertex_indexes.items():
        nk = vertex_renames.get(k, k)
        new_vi.setdefault(nk, []).extend(list(vlist))
    profile.vertex_indexes = new_vi

    profile.edge_specs = [
        s.model_copy(
            update={
                "source": vertex_renames.get(s.source, s.source),
                "target": vertex_renames.get(s.target, s.target),
            },
            deep=True,
        )
        for s in profile.edge_specs
    ]

    new_vpn: dict[str, dict[str, str]] = {}
    for k, names in profile.vertex_property_names.items():
        nk = vertex_renames.get(k, k)
        if nk in new_vpn and new_vpn[nk] != names:
            raise ValueError(
                f"Conflicting vertex_property_names for logical {nk!r}: "
                f"{new_vpn[nk]!r} vs {names!r}"
            )
        new_vpn[nk] = dict(names)
    profile.vertex_property_names = new_vpn

    dpv = profile.default_property_values
    if dpv is None:
        return

    new_vertices: dict[str, dict[str, Any]] = {}
    for k, props in dpv.vertices.items():
        nk = vertex_renames.get(k, k)
        if nk in new_vertices and new_vertices[nk] != props:
            raise ValueError(
                f"Conflicting default_property_values.vertices for logical {nk!r}"
            )
        new_vertices[nk] = dict(props)
    object.__setattr__(dpv, "vertices", new_vertices)

    new_edges: list[EdgePropertyDefaults] = []
    for e in dpv.edges:
        new_edges.append(
            EdgePropertyDefaults(
                source=vertex_renames.get(e.source, e.source),
                target=vertex_renames.get(e.target, e.target),
                relation=e.relation,
                values=dict(e.values),
            )
        )
    object.__setattr__(dpv, "edges", new_edges)

merge_relation_entries_in_db_profile(profile)

Merge duplicate edge-spec/default entries created by relation remaps.

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

def merge_relation_entries_in_db_profile(profile: DatabaseProfile) -> None:
    """Merge duplicate edge-spec/default entries created by relation remaps."""
    merged_specs: dict[EdgePhysicalKey, EdgePhysicalSpec] = {}
    for spec in profile.edge_specs:
        key = spec.physical_key
        current = merged_specs.get(key)
        if current is None:
            merged_specs[key] = spec
            continue
        if current.relation_name != spec.relation_name:
            raise ValueError(
                "Conflicting relation_name while merging relation entries for "
                f"{key}: {current.relation_name!r} vs {spec.relation_name!r}"
            )
        if current.indexes_mode != spec.indexes_mode:
            raise ValueError(
                "Conflicting indexes_mode while merging relation entries for "
                f"{key}: {current.indexes_mode!r} vs {spec.indexes_mode!r}"
            )
        merged_specs[key] = current.model_copy(
            update={
                "indexes": list(current.indexes) + list(spec.indexes),
                "property_names": _merge_vertex_default_maps(
                    dict(current.property_names),
                    dict(spec.property_names),
                    label=f"property_names while merging relation entries for {key}",
                ),
            },
            deep=True,
        )
    profile.edge_specs = list(merged_specs.values())

    dpv = profile.default_property_values
    if dpv is None:
        return
    merged_defaults: dict[EdgeId, EdgePropertyDefaults] = {}
    for edge in dpv.edges:
        current = merged_defaults.get(edge.edge_id)
        if current is None:
            merged_defaults[edge.edge_id] = edge
            continue
        merged_values = _merge_vertex_default_maps(
            dict(current.values),
            dict(edge.values),
            label="default_property_values.edges",
        )
        merged_defaults[edge.edge_id] = current.model_copy(
            update={"values": merged_values}, deep=True
        )
    object.__setattr__(dpv, "edges", list(merged_defaults.values()))

remap_vertices_in_db_profile(profile, mapping)

Remap logical vertex keys in profile by mapping, in one pass.

Every key is looked up once against its original name, so a chain or a swap resolves without an intermediate state; keys landing on one name merge (index lists concatenate, default maps union, storage names must agree).

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

def remap_vertices_in_db_profile(
    profile: DatabaseProfile, mapping: dict[str, str]
) -> None:
    """Remap logical vertex keys in *profile* by *mapping*, in one pass.

    Every key is looked up once against its original name, so a chain or a
    swap resolves without an intermediate state; keys landing on one name
    merge (index lists concatenate, default maps union, storage names must
    agree).
    """
    m = {k: v for k, v in mapping.items() if k != v}
    if not m:
        return

    new_vs: dict[str, str] = {}
    for k, v in profile.vertex_storage_names.items():
        nk = m.get(k, k)
        if nk in new_vs and new_vs[nk] != v:
            raise ValueError(
                f"Conflicting vertex_storage_names for logical {nk!r}: "
                f"{new_vs[nk]!r} vs {v!r}"
            )
        new_vs[nk] = v
    profile.vertex_storage_names = new_vs

    new_vi: dict[str, list[Any]] = {}
    for k, vlist in profile.vertex_indexes.items():
        nk = m.get(k, k)
        new_vi.setdefault(nk, []).extend(list(vlist))
    profile.vertex_indexes = new_vi

    profile.edge_specs = [
        s.model_copy(
            update={
                "source": m.get(s.source, s.source),
                "target": m.get(s.target, s.target),
            },
            deep=True,
        )
        for s in profile.edge_specs
    ]

    # Merged vertices pool their properties, so their physical names union;
    # two different physical names for one logical property cannot both hold.
    merged_vpn: dict[str, dict[str, str]] = {}
    for k, names in profile.vertex_property_names.items():
        nk = m.get(k, k)
        merged_vpn[nk] = _merge_vertex_default_maps(
            merged_vpn.get(nk, {}),
            dict(names),
            label="vertex_property_names",
        )
    profile.vertex_property_names = merged_vpn

    dpv = profile.default_property_values
    if dpv is None:
        return

    merged_vert: dict[str, dict[str, Any]] = {}
    for k, props in dpv.vertices.items():
        nk = m.get(k, k)
        if nk not in merged_vert:
            merged_vert[nk] = dict(props)
        else:
            merged_vert[nk] = _merge_vertex_default_maps(
                merged_vert[nk],
                dict(props),
                label="default_property_values.vertices",
            )
    object.__setattr__(dpv, "vertices", merged_vert)

    new_edges: list[EdgePropertyDefaults] = []
    for e in dpv.edges:
        src = m.get(e.source, e.source)
        tgt = m.get(e.target, e.target)
        new_edges.append(
            EdgePropertyDefaults(
                source=src,
                target=tgt,
                relation=e.relation,
                values=dict(e.values),
            )
        )
    object.__setattr__(dpv, "edges", new_edges)

retain_native_inverses(profile, relations)

Keep only the native inverses of relations.

A native inverse is keyed by relation, so removing edges drops it only once the relation has no edge left; callers pass the relations that survive.

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

def retain_native_inverses(profile: DatabaseProfile, relations: set[str]) -> None:
    """Keep only the native inverses of ``relations``.

    A native inverse is keyed by relation, so removing edges drops it only once
    the relation has no edge left; callers pass the relations that survive.
    """
    kept = [relation for relation in profile.native_inverses if relation in relations]
    if kept != profile.native_inverses:
        profile.native_inverses = kept

union_default_property_values(left, right)

Union two default-value blocks, refusing two defaults for one property.

A default is a physical DDL claim about one attribute of one type, so two sides supplying different ones is a contradiction rather than an ordering: whichever the fold elected would silently decide what a column holds for every row written without that attribute.

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

def union_default_property_values(
    left: DefaultPropertyValues | None, right: DefaultPropertyValues | None
) -> DefaultPropertyValues | None:
    """Union two default-value blocks, refusing two defaults for one property.

    A default is a physical DDL claim about one attribute of one type, so two
    sides supplying different ones is a contradiction rather than an ordering:
    whichever the fold elected would silently decide what a column holds for
    every row written without that attribute.
    """
    if left is None or right is None:
        source = left if right is None else right
        return source.model_copy(deep=True) if source is not None else None

    vertices: dict[str, dict[str, Any]] = {
        name: dict(values) for name, values in left.vertices.items()
    }
    for name, values in right.vertices.items():
        if name not in vertices:
            vertices[name] = dict(values)
            continue
        vertices[name] = _merge_vertex_default_maps(
            vertices[name],
            dict(values),
            label=f"default_property_values on vertex {name!r}",
        )

    edges: dict[EdgeId, EdgePropertyDefaults] = {}
    for entry in list(left.edges) + list(right.edges):
        current = edges.get(entry.edge_id)
        if current is None:
            edges[entry.edge_id] = entry.model_copy(deep=True)
            continue
        edges[entry.edge_id] = current.model_copy(
            update={
                "values": _merge_vertex_default_maps(
                    dict(current.values),
                    dict(entry.values),
                    label=f"default_property_values on edge {entry.edge_id!r}",
                )
            },
            deep=True,
        )

    return DefaultPropertyValues(vertices=vertices, edges=list(edges.values()))